Repository: databendlabs/databend-sqlalchemy (language: python)

Issue:
Limit & Offset do not work correctly
Small script to replicate limit clause that does not execute against database.
```python
import sqlalchemy as sa
metadata = sa.MetaData()
tbl = sa.Table(
    "test-table",
    metadata,
    sa.Column("Col1", sa.DECIMAL(38, 10)),
    sa.Column("Col2", sa.VARCHAR()),
    sa.Column("Col3", sa.VARCHAR()),
    schema="test-schema",
)
select_query = sa.select(tbl).limit(1000).offset(0)
engine = sa.create_engine("databend://root:@localhost:8000?sslmode=disable")

print(select_query.compile(engine, compile_kwargs={"literal_binds": True, "render_postcompile": True}))
```
Produces this output which will not work
```sql
SELECT "Col1", "Col2", "Col3" 
FROM "test-schema"."test-table"
 LIMIT 0,0
```

Expecting this or similar (something with limit of 1000)
```sql
SELECT "Col1", "Col2", "Col3" 
FROM "test-schema"."test-table"
 LIMIT 1000 OFFSET 0
```

Also relates to https://github.com/datafuselabs/databend-sqlalchemy/issues/26 where parameter substitution does not happen

Parameter Substitution not happening for SQL statement
Given a simple select that has a limit/offset clause, parameter substitution does not occur for the SQL during execution so it fails to execute.

```python
import sqlalchemy as sa

metadata = sa.MetaData()

tbl = sa.Table(
    "test-table",
    metadata,
    sa.Column("Col1", sa.DECIMAL(38, 10)),
    sa.Column("Col2", sa.VARCHAR()),
    sa.Column("Col3", sa.VARCHAR()),
    schema="test-schema",
)
select_query = sa.select(tbl).limit(1000).offset(0)

engine = sa.create_engine("databend://root:@plaid-databend-query:8000?sslmode=disable")

# print(select_query.compile(engine, compile_kwargs={"literal_binds": True, "render_postcompile": True}))
with engine.connect() as conn:
    conn.execute(select_query).fetchall()
```

Actual SQL generated by SQLAlchemy normal compilation (although incorrect values for limit/offset)
```sql
SELECT "Col1", "Col2", "Col3" 
FROM "test-schema"."test-table"
 LIMIT 0,0
```

```bash
Traceback (most recent call last):
  File "/home/plaid/src/plaid/plaid/temp/patbuxton/databend_sql.py", line 22, in <module>
    conn.execute(select_query).fetchall()
    ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1385, in execute
    return meth(self, multiparams, params, _EMPTY_EXECUTION_OPTS)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/sql/elements.py", line 334, in _execute_on_connection
    return connection._execute_clauseelement(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1577, in _execute_clauseelement
    ret = self._execute_context(
          ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1953, in _execute_context
    self._handle_dbapi_exception(
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 2138, in _handle_dbapi_exception
    util.raise_(exc_info[1], with_traceback=exc_info[2])
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py", line 1910, in _execute_context
    self.dialect.do_execute(
  File "/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/default.py", line 736, in do_execute
    cursor.execute(statement, parameters)
  File "/usr/local/lib/python3.11/site-packages/databend_sqlalchemy/connector.py", line 182, in execute
    rows = self._db.query_iter(operation)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Exception: APIError: ResponseError with 1005: error: 
  --> SQL:3:10
  |
1 | SELECT "Col1", "Col2", "Col3" 
  | ------ while parsing `SELECT ...`
2 | FROM "test-schema"."test-table"
3 |  LIMIT 0,%(param_2)s
  |          ^
  |          |
  |          missing lhs or rhs for the binary operator
  |          while parsing expression
```


Discussion / hints:

Fix patch:
diff --git a/Makefile b/Makefile
index b85531a..294a30f 100644
--- a/Makefile
+++ b/Makefile
@@ -4,8 +4,8 @@ unittest:
 integration:
 	python -m pytest -s tests/integration
 
-sampletest:
-	python databend_sqlalchemy/test.py
+testsuite:
+    python -m pytest -n4
 
 install:
 	pip install -e ".[dev]"
diff --git a/Pipfile b/Pipfile
index 3d187fb..bbd4705 100644
--- a/Pipfile
+++ b/Pipfile
@@ -10,6 +10,8 @@ databend-sqlalchemy = {file = ".", editable = true}
 black = "*"
 flake8 = "*"
 pytest = "*"
+pytest-xdist = "*"
+
 
 [requires]
 python_version = "3.11"
diff --git a/Pipfile.lock b/Pipfile.lock
index f4d7e5d..33be966 100644
--- a/Pipfile.lock
+++ b/Pipfile.lock
@@ -1,7 +1,7 @@
 {
     "_meta": {
         "hash": {
-            "sha256": "ca24d636b33804912b154c29eec6e6e2d3151d4cbd8f3c51d189082e4c9bc1b4"
+            "sha256": "67cff3649440470951c241bf14d15e31c6a99e02ba0dc2c7dd07e36431964693"
         },
         "pipfile-spec": 6,
         "requires": {
@@ -18,111 +18,176 @@
     "default": {
         "databend-driver": {
             "hashes": [
-                "sha256:46c9e78454fc45c54ca325cee1427ae52f560150466e13a2c69054647a881d44",
-                "sha256:7721ddb4721da899afc614c145ceecd79c90a31984d97b7195c1df6eb46ea5b7",
-                "sha256:a575d8f452fa1c7ac6b2f96c3dec86880fc9359eb75a1a4bdbc5ffefb0460b7d",
-                "sha256:d25f457aa119d97b5135bf05311db6cf625680766dd5fe12034c0b8c483238e1"
+                "sha256:2502a5393d568531b3a52685bf769ecf573cdd5df12bc43d3e8dd728ae74bd12",
+                "sha256:62acf48a7ddf6a1f9ff589dcca4b15d0701f268b814587824e1f71f261121375",
+                "sha256:c63f45a21e90b3b08ce2d8f18c3b6cf2b3f122f2a7f0e1868b868751632fd679",
+                "sha256:eb6d64be270dfc5581dcab9c139336f73429bfe2624175523db2d7adac7d3c45"
             ],
             "markers": "python_version >= '3.7'",
-            "version": "==0.12.3"
+            "version": "==0.16.0"
         },
         "databend-sqlalchemy": {
             "editable": true,
-            "file": "."
+            "file": ".",
+            "markers": "python_version >= '3.7'"
+        },
+        "greenlet": {
+            "hashes": [
+                "sha256:01bc7ea167cf943b4c802068e178bbf70ae2e8c080467070d01bfa02f337ee67",
+                "sha256:0448abc479fab28b00cb472d278828b3ccca164531daab4e970a0458786055d6",
+                "sha256:086152f8fbc5955df88382e8a75984e2bb1c892ad2e3c80a2508954e52295257",
+                "sha256:098d86f528c855ead3479afe84b49242e174ed262456c342d70fc7f972bc13c4",
+                "sha256:149e94a2dd82d19838fe4b2259f1b6b9957d5ba1b25640d2380bea9c5df37676",
+                "sha256:1551a8195c0d4a68fac7a4325efac0d541b48def35feb49d803674ac32582f61",
+                "sha256:15d79dd26056573940fcb8c7413d84118086f2ec1a8acdfa854631084393efcc",
+                "sha256:1996cb9306c8595335bb157d133daf5cf9f693ef413e7673cb07e3e5871379ca",
+                "sha256:1a7191e42732df52cb5f39d3527217e7ab73cae2cb3694d241e18f53d84ea9a7",
+                "sha256:1ea188d4f49089fc6fb283845ab18a2518d279c7cd9da1065d7a84e991748728",
+                "sha256:1f672519db1796ca0d8753f9e78ec02355e862d0998193038c7073045899f305",
+                "sha256:2516a9957eed41dd8f1ec0c604f1cdc86758b587d964668b5b196a9db5bfcde6",
+                "sha256:2797aa5aedac23af156bbb5a6aa2cd3427ada2972c828244eb7d1b9255846379",
+                "sha256:2dd6e660effd852586b6a8478a1d244b8dc90ab5b1321751d2ea15deb49ed414",
+                "sha256:3ddc0f794e6ad661e321caa8d2f0a55ce01213c74722587256fb6566049a8b04",
+                "sha256:3ed7fb269f15dc662787f4119ec300ad0702fa1b19d2135a37c2c4de6fadfd4a",
+                "sha256:419b386f84949bf0e7c73e6032e3457b82a787c1ab4a0e43732898a761cc9dbf",
+                "sha256:43374442353259554ce33599da8b692d5aa96f8976d567d4badf263371fbe491",
+                "sha256:52f59dd9c96ad2fc0d5724107444f76eb20aaccb675bf825df6435acb7703559",
+                "sha256:57e8974f23e47dac22b83436bdcf23080ade568ce77df33159e019d161ce1d1e",
+                "sha256:5b51e85cb5ceda94e79d019ed36b35386e8c37d22f07d6a751cb659b180d5274",
+                "sha256:649dde7de1a5eceb258f9cb00bdf50e978c9db1b996964cd80703614c86495eb",
+                "sha256:64d7675ad83578e3fc149b617a444fab8efdafc9385471f868eb5ff83e446b8b",
+                "sha256:68834da854554926fbedd38c76e60c4a2e3198c6fbed520b106a8986445caaf9",
+                "sha256:6b66c9c1e7ccabad3a7d037b2bcb740122a7b17a53734b7d72a344ce39882a1b",
+                "sha256:70fb482fdf2c707765ab5f0b6655e9cfcf3780d8d87355a063547b41177599be",
+                "sha256:7170375bcc99f1a2fbd9c306f5be8764eaf3ac6b5cb968862cad4c7057756506",
+                "sha256:73a411ef564e0e097dbe7e866bb2dda0f027e072b04da387282b02c308807405",
+                "sha256:77457465d89b8263bca14759d7c1684df840b6811b2499838cc5b040a8b5b113",
+                "sha256:7f362975f2d179f9e26928c5b517524e89dd48530a0202570d55ad6ca5d8a56f",
+                "sha256:81bb9c6d52e8321f09c3d165b2a78c680506d9af285bfccbad9fb7ad5a5da3e5",
+                "sha256:881b7db1ebff4ba09aaaeae6aa491daeb226c8150fc20e836ad00041bcb11230",
+                "sha256:894393ce10ceac937e56ec00bb71c4c2f8209ad516e96033e4b3b1de270e200d",
+                "sha256:99bf650dc5d69546e076f413a87481ee1d2d09aaaaaca058c9251b6d8c14783f",
+                "sha256:9da2bd29ed9e4f15955dd1595ad7bc9320308a3b766ef7f837e23ad4b4aac31a",
+                "sha256:afaff6cf5200befd5cec055b07d1c0a5a06c040fe5ad148abcd11ba6ab9b114e",
+                "sha256:b1b5667cced97081bf57b8fa1d6bfca67814b0afd38208d52538316e9422fc61",
+                "sha256:b37eef18ea55f2ffd8f00ff8fe7c8d3818abd3e25fb73fae2ca3b672e333a7a6",
+                "sha256:b542be2440edc2d48547b5923c408cbe0fc94afb9f18741faa6ae970dbcb9b6d",
+                "sha256:b7dcbe92cc99f08c8dd11f930de4d99ef756c3591a5377d1d9cd7dd5e896da71",
+                "sha256:b7f009caad047246ed379e1c4dbcb8b020f0a390667ea74d2387be2998f58a22",
+                "sha256:bba5387a6975598857d86de9eac14210a49d554a77eb8261cc68b7d082f78ce2",
+                "sha256:c5e1536de2aad7bf62e27baf79225d0d64360d4168cf2e6becb91baf1ed074f3",
+                "sha256:c5ee858cfe08f34712f548c3c363e807e7186f03ad7a5039ebadb29e8c6be067",
+                "sha256:c9db1c18f0eaad2f804728c67d6c610778456e3e1cc4ab4bbd5eeb8e6053c6fc",
+                "sha256:d353cadd6083fdb056bb46ed07e4340b0869c305c8ca54ef9da3421acbdf6881",
+                "sha256:d46677c85c5ba00a9cb6f7a00b2bfa6f812192d2c9f7d9c4f6a55b60216712f3",
+                "sha256:d4d1ac74f5c0c0524e4a24335350edad7e5f03b9532da7ea4d3c54d527784f2e",
+                "sha256:d73a9fe764d77f87f8ec26a0c85144d6a951a6c438dfe50487df5595c6373eac",
+                "sha256:da70d4d51c8b306bb7a031d5cff6cc25ad253affe89b70352af5f1cb68e74b53",
+                "sha256:daf3cb43b7cf2ba96d614252ce1684c1bccee6b2183a01328c98d36fcd7d5cb0",
+                "sha256:dca1e2f3ca00b84a396bc1bce13dd21f680f035314d2379c4160c98153b2059b",
+                "sha256:dd4f49ae60e10adbc94b45c0b5e6a179acc1736cf7a90160b404076ee283cf83",
+                "sha256:e1f145462f1fa6e4a4ae3c0f782e580ce44d57c8f2c7aae1b6fa88c0b2efdb41",
+                "sha256:e3391d1e16e2a5a1507d83e4a8b100f4ee626e8eca43cf2cadb543de69827c4c",
+                "sha256:fcd2469d6a2cf298f198f0487e0a5b1a47a42ca0fa4dfd1b6862c999f018ebbf",
+                "sha256:fd096eb7ffef17c456cfa587523c5f92321ae02427ff955bebe9e3c63bc9f0da",
+                "sha256:fe754d231288e1e64323cfad462fcee8f0288654c10bdf4f603a39ed923bef33"
+            ],
+            "markers": "platform_machine == 'aarch64' or (platform_machine == 'ppc64le' or (platform_machine == 'x86_64' or (platform_machine == 'amd64' or (platform_machine == 'AMD64' or (platform_machine == 'win32' or platform_machine == 'WIN32')))))",
+            "version": "==3.0.3"
         },
         "sqlalchemy": {
             "hashes": [
-                "sha256:0d3cab3076af2e4aa5693f89622bef7fa770c6fec967143e4da7508b3dceb9b9",
-                "sha256:0dacf67aee53b16f365c589ce72e766efaabd2b145f9de7c917777b575e3659d",
-                "sha256:10331f129982a19df4284ceac6fe87353ca3ca6b4ca77ff7d697209ae0a5915e",
-                "sha256:14a6f68e8fc96e5e8f5647ef6cda6250c780612a573d99e4d881581432ef1669",
-                "sha256:1b1180cda6df7af84fe72e4530f192231b1f29a7496951db4ff38dac1687202d",
-                "sha256:29049e2c299b5ace92cbed0c1610a7a236f3baf4c6b66eb9547c01179f638ec5",
-                "sha256:342d365988ba88ada8af320d43df4e0b13a694dbd75951f537b2d5e4cb5cd002",
-                "sha256:420362338681eec03f53467804541a854617faed7272fe71a1bfdb07336a381e",
-                "sha256:4344d059265cc8b1b1be351bfb88749294b87a8b2bbe21dfbe066c4199541ebd",
-                "sha256:4f7a7d7fcc675d3d85fbf3b3828ecd5990b8d61bd6de3f1b260080b3beccf215",
-                "sha256:555651adbb503ac7f4cb35834c5e4ae0819aab2cd24857a123370764dc7d7e24",
-                "sha256:59a21853f5daeb50412d459cfb13cb82c089ad4c04ec208cd14dddd99fc23b39",
-                "sha256:5fdd402169aa00df3142149940b3bf9ce7dde075928c1886d9a1df63d4b8de62",
-                "sha256:605b6b059f4b57b277f75ace81cc5bc6335efcbcc4ccb9066695e515dbdb3900",
-                "sha256:665f0a3954635b5b777a55111ababf44b4fc12b1f3ba0a435b602b6387ffd7cf",
-                "sha256:6f9e2e59cbcc6ba1488404aad43de005d05ca56e069477b33ff74e91b6319735",
-                "sha256:736ea78cd06de6c21ecba7416499e7236a22374561493b456a1f7ffbe3f6cdb4",
-                "sha256:74b080c897563f81062b74e44f5a72fa44c2b373741a9ade701d5f789a10ba23",
-                "sha256:75432b5b14dc2fff43c50435e248b45c7cdadef73388e5610852b95280ffd0e9",
-                "sha256:75f99202324383d613ddd1f7455ac908dca9c2dd729ec8584c9541dd41822a2c",
-                "sha256:790f533fa5c8901a62b6fef5811d48980adeb2f51f1290ade8b5e7ba990ba3de",
-                "sha256:798f717ae7c806d67145f6ae94dc7c342d3222d3b9a311a784f371a4333212c7",
-                "sha256:7c88f0c7dcc5f99bdb34b4fd9b69b93c89f893f454f40219fe923a3a2fd11625",
-                "sha256:7d505815ac340568fd03f719446a589162d55c52f08abd77ba8964fbb7eb5b5f",
-                "sha256:84daa0a2055df9ca0f148a64fdde12ac635e30edbca80e87df9b3aaf419e144a",
-                "sha256:87d91043ea0dc65ee583026cb18e1b458d8ec5fc0a93637126b5fc0bc3ea68c4",
-                "sha256:87f6e732bccd7dcf1741c00f1ecf33797383128bd1c90144ac8adc02cbb98643",
-                "sha256:884272dcd3ad97f47702965a0e902b540541890f468d24bd1d98bcfe41c3f018",
-                "sha256:8b8cb63d3ea63b29074dcd29da4dc6a97ad1349151f2d2949495418fd6e48db9",
-                "sha256:91f7d9d1c4dd1f4f6e092874c128c11165eafcf7c963128f79e28f8445de82d5",
-                "sha256:a2c69a7664fb2d54b8682dd774c3b54f67f84fa123cf84dda2a5f40dcaa04e08",
-                "sha256:a3be4987e3ee9d9a380b66393b77a4cd6d742480c951a1c56a23c335caca4ce3",
-                "sha256:a86b4240e67d4753dc3092d9511886795b3c2852abe599cffe108952f7af7ac3",
-                "sha256:aa9373708763ef46782d10e950b49d0235bfe58facebd76917d3f5cbf5971aed",
-                "sha256:b64b183d610b424a160b0d4d880995e935208fc043d0302dd29fee32d1ee3f95",
-                "sha256:b801154027107461ee992ff4b5c09aa7cc6ec91ddfe50d02bca344918c3265c6",
-                "sha256:bb209a73b8307f8fe4fe46f6ad5979649be01607f11af1eb94aa9e8a3aaf77f0",
-                "sha256:bc8b7dabe8e67c4832891a5d322cec6d44ef02f432b4588390017f5cec186a84",
-                "sha256:c51db269513917394faec5e5c00d6f83829742ba62e2ac4fa5c98d58be91662f",
-                "sha256:c55731c116806836a5d678a70c84cb13f2cedba920212ba7dcad53260997666d",
-                "sha256:cf18ff7fc9941b8fc23437cc3e68ed4ebeff3599eec6ef5eebf305f3d2e9a7c2",
-                "sha256:d24f571990c05f6b36a396218f251f3e0dda916e0c687ef6fdca5072743208f5",
-                "sha256:db854730a25db7c956423bb9fb4bdd1216c839a689bf9cc15fada0a7fb2f4570",
-                "sha256:dc55990143cbd853a5d038c05e79284baedf3e299661389654551bd02a6a68d7",
-                "sha256:e607cdd99cbf9bb80391f54446b86e16eea6ad309361942bf88318bcd452363c",
-                "sha256:ecf6d4cda1f9f6cb0b45803a01ea7f034e2f1aed9475e883410812d9f9e3cfcf",
-                "sha256:f2a159111a0f58fb034c93eeba211b4141137ec4b0a6e75789ab7a3ef3c7e7e3",
-                "sha256:f37c0caf14b9e9b9e8f6dbc81bc56db06acb4363eba5a633167781a48ef036ed",
-                "sha256:f5693145220517b5f42393e07a6898acdfe820e136c98663b971906120549da5"
+                "sha256:01d10638a37460616708062a40c7b55f73e4d35eaa146781c683e0fa7f6c43fb",
+                "sha256:04c487305ab035a9548f573763915189fc0fe0824d9ba28433196f8436f1449c",
+                "sha256:0dfefdb3e54cd15f5d56fd5ae32f1da2d95d78319c1f6dfb9bcd0eb15d603d5d",
+                "sha256:0f3ca96af060a5250a8ad5a63699180bc780c2edf8abf96c58af175921df847a",
+                "sha256:205f5a2b39d7c380cbc3b5dcc8f2762fb5bcb716838e2d26ccbc54330775b003",
+                "sha256:25664e18bef6dc45015b08f99c63952a53a0a61f61f2e48a9e70cec27e55f699",
+                "sha256:296195df68326a48385e7a96e877bc19aa210e485fa381c5246bc0234c36c78e",
+                "sha256:2a0732dffe32333211801b28339d2a0babc1971bc90a983e3035e7b0d6f06b93",
+                "sha256:3071ad498896907a5ef756206b9dc750f8e57352113c19272bdfdc429c7bd7de",
+                "sha256:308ef9cb41d099099fffc9d35781638986870b29f744382904bf9c7dadd08513",
+                "sha256:334184d1ab8f4c87f9652b048af3f7abea1c809dfe526fb0435348a6fef3d380",
+                "sha256:38b624e5cf02a69b113c8047cf7f66b5dfe4a2ca07ff8b8716da4f1b3ae81567",
+                "sha256:471fcb39c6adf37f820350c28aac4a7df9d3940c6548b624a642852e727ea586",
+                "sha256:4c142852ae192e9fe5aad5c350ea6befe9db14370b34047e1f0f7cf99e63c63b",
+                "sha256:4f6d971255d9ddbd3189e2e79d743ff4845c07f0633adfd1de3f63d930dbe673",
+                "sha256:52c8011088305476691b8750c60e03b87910a123cfd9ad48576d6414b6ec2a1d",
+                "sha256:52de4736404e53c5c6a91ef2698c01e52333988ebdc218f14c833237a0804f1b",
+                "sha256:5c7b02525ede2a164c5fa5014915ba3591730f2cc831f5be9ff3b7fd3e30958e",
+                "sha256:5ef3fbccb4058355053c51b82fd3501a6e13dd808c8d8cd2561e610c5456013c",
+                "sha256:5f20cb0a63a3e0ec4e169aa8890e32b949c8145983afa13a708bc4b0a1f30e03",
+                "sha256:61405ea2d563407d316c63a7b5271ae5d274a2a9fbcd01b0aa5503635699fa1e",
+                "sha256:77d29cb6c34b14af8a484e831ab530c0f7188f8efed1c6a833a2c674bf3c26ec",
+                "sha256:7b184e3de58009cc0bf32e20f137f1ec75a32470f5fede06c58f6c355ed42a72",
+                "sha256:7e614d7a25a43a9f54fcce4675c12761b248547f3d41b195e8010ca7297c369c",
+                "sha256:8197d6f7a3d2b468861ebb4c9f998b9df9e358d6e1cf9c2a01061cb9b6cf4e41",
+                "sha256:87a1d53a5382cdbbf4b7619f107cc862c1b0a4feb29000922db72e5a66a5ffc0",
+                "sha256:8c37f1050feb91f3d6c32f864d8e114ff5545a4a7afe56778d76a9aec62638ba",
+                "sha256:90453597a753322d6aa770c5935887ab1fc49cc4c4fdd436901308383d698b4b",
+                "sha256:988569c8732f54ad3234cf9c561364221a9e943b78dc7a4aaf35ccc2265f1930",
+                "sha256:99a1e69d4e26f71e750e9ad6fdc8614fbddb67cfe2173a3628a2566034e223c7",
+                "sha256:9b19836ccca0d321e237560e475fd99c3d8655d03da80c845c4da20dda31b6e1",
+                "sha256:9d6753305936eddc8ed190e006b7bb33a8f50b9854823485eed3a886857ab8d1",
+                "sha256:a13b917b4ffe5a0a31b83d051d60477819ddf18276852ea68037a144a506efb9",
+                "sha256:a88913000da9205b13f6f195f0813b6ffd8a0c0c2bd58d499e00a30eb508870c",
+                "sha256:b2a0e3cf0caac2085ff172c3faacd1e00c376e6884b5bc4dd5b6b84623e29e4f",
+                "sha256:b5d7ed79df55a731749ce65ec20d666d82b185fa4898430b17cb90c892741520",
+                "sha256:bab41acf151cd68bc2b466deae5deeb9e8ae9c50ad113444151ad965d5bf685b",
+                "sha256:bd9566b8e58cabd700bc367b60e90d9349cd16f0984973f98a9a09f9c64e86f0",
+                "sha256:bda7ce59b06d0f09afe22c56714c65c957b1068dee3d5e74d743edec7daba552",
+                "sha256:c2f9c762a2735600654c654bf48dad388b888f8ce387b095806480e6e4ff6907",
+                "sha256:c4520047006b1d3f0d89e0532978c0688219857eb2fee7c48052560ae76aca1e",
+                "sha256:d96710d834a6fb31e21381c6d7b76ec729bd08c75a25a5184b1089141356171f",
+                "sha256:dba622396a3170974f81bad49aacebd243455ec3cc70615aeaef9e9613b5bca5",
+                "sha256:dc4ee2d4ee43251905f88637d5281a8d52e916a021384ec10758826f5cbae305",
+                "sha256:dddaae9b81c88083e6437de95c41e86823d150f4ee94bf24e158a4526cbead01",
+                "sha256:de7202ffe4d4a8c1e3cde1c03e01c1a3772c92858837e8f3879b497158e4cb44",
+                "sha256:e5bbe55e8552019c6463709b39634a5fc55e080d0827e2a3a11e18eb73f5cdbd",
+                "sha256:ea311d4ee9a8fa67f139c088ae9f905fcf0277d6cd75c310a21a88bf85e130f5",
+                "sha256:fecd5089c4be1bcc37c35e9aa678938d2888845a134dd016de457b942cf5a758"
             ],
             "markers": "python_version >= '3.7'",
-            "version": "==2.0.25"
+            "version": "==2.0.29"
         },
         "typing-extensions": {
             "hashes": [
-                "sha256:23478f88c37f27d76ac8aee6c905017a143b0b1b886c3c9f66bc2fd94f9f5783",
-                "sha256:af72aea155e91adfc61c3ae9e0e342dbc0cba726d6cba4b6c72c1f34e47291cd"
+                "sha256:69b1a937c3a517342112fb4c6df7e72fc39a38e7891a5730ed4985b5214b5475",
+                "sha256:b0abd7c89e8fb96f98db18d86106ff1d90ab692004eb746cf6eda2682f91b3cb"
             ],
             "markers": "python_version >= '3.8'",
-            "version": "==4.9.0"
+            "version": "==4.10.0"
         }
     },
     "develop": {
         "black": {
             "hashes": [
-                "sha256:0808494f2b2df923ffc5723ed3c7b096bd76341f6213989759287611e9837d50",
-                "sha256:1fa88a0f74e50e4487477bc0bb900c6781dbddfdfa32691e780bf854c3b4a47f",
-                "sha256:25e57fd232a6d6ff3f4478a6fd0580838e47c93c83eaf1ccc92d4faf27112c4e",
-                "sha256:2d9e13db441c509a3763a7a3d9a49ccc1b4e974a47be4e08ade2a228876500ec",
-                "sha256:3e1b38b3135fd4c025c28c55ddfc236b05af657828a8a6abe5deec419a0b7055",
-                "sha256:3fa4be75ef2a6b96ea8d92b1587dd8cb3a35c7e3d51f0738ced0781c3aa3a5a3",
-                "sha256:4ce3ef14ebe8d9509188014d96af1c456a910d5b5cbf434a09fef7e024b3d0d5",
-                "sha256:4f0031eaa7b921db76decd73636ef3a12c942ed367d8c3841a0739412b260a54",
-                "sha256:602cfb1196dc692424c70b6507593a2b29aac0547c1be9a1d1365f0d964c353b",
-                "sha256:6d1bd9c210f8b109b1762ec9fd36592fdd528485aadb3f5849b2740ef17e674e",
-                "sha256:78baad24af0f033958cad29731e27363183e140962595def56423e626f4bee3e",
-                "sha256:8d4df77958a622f9b5a4c96edb4b8c0034f8434032ab11077ec6c56ae9f384ba",
-                "sha256:97e56155c6b737854e60a9ab1c598ff2533d57e7506d97af5481141671abf3ea",
-                "sha256:9c4352800f14be5b4864016882cdba10755bd50805c95f728011bcb47a4afd59",
-                "sha256:a4d6a9668e45ad99d2f8ec70d5c8c04ef4f32f648ef39048d010b0689832ec6d",
-                "sha256:a920b569dc6b3472513ba6ddea21f440d4b4c699494d2e972a1753cdc25df7b0",
-                "sha256:ae76c22bde5cbb6bfd211ec343ded2163bba7883c7bc77f6b756a1049436fbb9",
-                "sha256:b18fb2ae6c4bb63eebe5be6bd869ba2f14fd0259bda7d18a46b764d8fb86298a",
-                "sha256:c04b6d9d20e9c13f43eee8ea87d44156b8505ca8a3c878773f68b4e4812a421e",
-                "sha256:c88b3711d12905b74206227109272673edce0cb29f27e1385f33b0163c414bba",
-                "sha256:dd15245c8b68fe2b6bd0f32c1556509d11bb33aec9b5d0866dd8e2ed3dba09c2",
-                "sha256:e0aaf6041986767a5e0ce663c7a2f0e9eaf21e6ff87a5f95cbf3675bfd4c41d2"
+                "sha256:2818cf72dfd5d289e48f37ccfa08b460bf469e67fb7c4abb07edc2e9f16fb63f",
+                "sha256:41622020d7120e01d377f74249e677039d20e6344ff5851de8a10f11f513bf93",
+                "sha256:4acf672def7eb1725f41f38bf6bf425c8237248bb0804faa3965c036f7672d11",
+                "sha256:4be5bb28e090456adfc1255e03967fb67ca846a03be7aadf6249096100ee32d0",
+                "sha256:4f1373a7808a8f135b774039f61d59e4be7eb56b2513d3d2f02a8b9365b8a8a9",
+                "sha256:56f52cfbd3dabe2798d76dbdd299faa046a901041faf2cf33288bc4e6dae57b5",
+                "sha256:65b76c275e4c1c5ce6e9870911384bff5ca31ab63d19c76811cb1fb162678213",
+                "sha256:65c02e4ea2ae09d16314d30912a58ada9a5c4fdfedf9512d23326128ac08ac3d",
+                "sha256:6905238a754ceb7788a73f02b45637d820b2f5478b20fec82ea865e4f5d4d9f7",
+                "sha256:79dcf34b33e38ed1b17434693763301d7ccbd1c5860674a8f871bd15139e7837",
+                "sha256:7bb041dca0d784697af4646d3b62ba4a6b028276ae878e53f6b4f74ddd6db99f",
+                "sha256:7d5e026f8da0322b5662fa7a8e752b3fa2dac1c1cbc213c3d7ff9bdd0ab12395",
+                "sha256:9f50ea1132e2189d8dff0115ab75b65590a3e97de1e143795adb4ce317934995",
+                "sha256:a0c9c4a0771afc6919578cec71ce82a3e31e054904e7197deacbc9382671c41f",
+                "sha256:aadf7a02d947936ee418777e0247ea114f78aff0d0959461057cae8a04f20597",
+                "sha256:b5991d523eee14756f3c8d5df5231550ae8993e2286b8014e2fdea7156ed0959",
+                "sha256:bf21b7b230718a5f08bd32d5e4f1db7fc8788345c8aea1d155fc17852b3410f5",
+                "sha256:c45f8dff244b3c431b36e3224b6be4a127c6aca780853574c00faf99258041eb",
+                "sha256:c7ed6668cbbfcd231fa0dc1b137d3e40c04c7f786e626b405c62bcd5db5857e4",
+                "sha256:d7de8d330763c66663661a1ffd432274a2f92f07feeddd89ffd085b5744f85e7",
+                "sha256:e19cb1c6365fd6dc38a6eae2dcb691d7d83935c10215aef8e6c38edee3f77abd",
+                "sha256:e2af80566f43c85f5797365077fb64a393861a3730bd110971ab7a0c94e873e7"
             ],
             "index": "pypi",
             "markers": "python_version >= '3.8'",
-            "version": "==23.12.1"
+            "version": "==24.3.0"
         },
         "click": {
             "hashes": [
@@ -132,6 +197,14 @@
             "markers": "python_version >= '3.7'",
             "version": "==8.1.7"
         },
+        "execnet": {
+            "hashes": [
+                "sha256:88256416ae766bc9e8895c76a87928c0012183da3cc4fc18016e6f050e025f41",
+                "sha256:cc59bc4423742fd71ad227122eb0dd44db51efb3dc4095b45ac9a08c770096af"
+            ],
+            "markers": "python_version >= '3.7'",
+            "version": "==2.0.2"
+        },
         "flake8": {
             "hashes": [
                 "sha256:33f96621059e65eec474169085dc92bf26e7b2d47366b70be2f67ab80dc25132",
@@ -167,11 +240,11 @@
         },
         "packaging": {
             "hashes": [
-                "sha256:048fb0e9405036518eaaf48a55953c750c11e1a1b68e0dd1a9d62ed0c092cfc5",
-                "sha256:8c491190033a9af7e1d931d0b5dacc2ef47509b34dd0de67ed209b5203fc88c7"
+                "sha256:2ddfb553fdf02fb784c234c7ba6ccc288296ceabec964ad2eae3777778130bc5",
+                "sha256:eb82c5e3e56209074766e6885bb04b8c38a0c015d0a30036ebe7ece34c9989e9"
             ],
             "markers": "python_version >= '3.7'",
-            "version": "==23.2"
+            "version": "==24.0"
         },
         "pathspec": {
             "hashes": [
@@ -183,19 +256,19 @@
         },
         "platformdirs": {
             "hashes": [
-                "sha256:11c8f37bcca40db96d8144522d925583bdb7a31f7b0e37e3ed4318400a8e2380",
-                "sha256:906d548203468492d432bcb294d4bc2fff751bf84971fbb2c10918cc206ee420"
+                "sha256:0614df2a2f37e1a662acbd8e2b25b92ccf8632929bc6d43467e17fe89c75e068",
+                "sha256:ef0cc731df711022c174543cb70a9b5bd22e5a9337c8624ef2c2ceb8ddad8768"
             ],
             "markers": "python_version >= '3.8'",
-            "version": "==4.1.0"
+            "version": "==4.2.0"
         },
         "pluggy": {
             "hashes": [
-                "sha256:cf61ae8f126ac6f7c451172cf30e3e43d3ca77615509771b3a984a0730651e12",
-                "sha256:d89c696a773f8bd377d18e5ecda92b7a3793cbe66c87060a6fb58c7b6e1061f7"
+                "sha256:7db9f7b503d67d1c5b95f59773ebb58a8c1c288129a88665838012cfb07b8981",
+                "sha256:8c85c2876142a764e5b7548e7d9a0e0ddb46f5185161049a79b7e974454223be"
             ],
             "markers": "python_version >= '3.8'",
-            "version": "==1.3.0"
+            "version": "==1.4.0"
         },
         "pycodestyle": {
             "hashes": [
@@ -215,12 +288,21 @@
         },
         "pytest": {
             "hashes": [
-                "sha256:2cf0005922c6ace4a3e2ec8b4080eb0d9753fdc93107415332f50ce9e7994280",
-                "sha256:b090cdf5ed60bf4c45261be03239c2c1c22df034fbffe691abe93cd80cea01d8"
+                "sha256:2a8386cfc11fa9d2c50ee7b2a57e7d898ef90470a7a34c4b949ff59662bb78b7",
+                "sha256:ac978141a75948948817d360297b7aae0fcb9d6ff6bc9ec6d514b85d5a65c044"
+            ],
+            "index": "pypi",
+            "markers": "python_version >= '3.8'",
+            "version": "==8.1.1"
+        },
+        "pytest-xdist": {
+            "hashes": [
+                "sha256:cbb36f3d67e0c478baa57fa4edc8843887e0f6cfc42d677530a36d7472b32d8a",
+                "sha256:d075629c7e00b611df89f490a5063944bee7a4362a5ff11c7cc7824a03dfce24"
             ],
             "index": "pypi",
             "markers": "python_version >= '3.7'",
-            "version": "==7.4.4"
+            "version": "==3.5.0"
         }
     }
 }
diff --git a/databend_sqlalchemy/connector.py b/databend_sqlalchemy/connector.py
index 02d4f65..b582100 100644
--- a/databend_sqlalchemy/connector.py
+++ b/databend_sqlalchemy/connector.py
@@ -3,11 +3,11 @@
 # See http://www.python.org/dev/peps/pep-0249/
 #
 # Many docstrings in this file are based on the PEP, which is in the public domain.
-
+import decimal
 import re
 import uuid
-from datetime import datetime
-from databend_sqlalchemy.errors import ServerException, NotSupportedError
+from datetime import datetime, date
+from databend_sqlalchemy.errors import Error, ServerException, NotSupportedError
 
 from databend_driver import BlockingDatabendClient
 
@@ -15,17 +15,10 @@
 apilevel = "2.0"
 threadsafety = 2  # Threads may share the module and connections.
 paramstyle = "pyformat"  # Python extended format codes, e.g. ...WHERE name=%(name)s
+Binary = bytes  # to satisfy dialect.dbapi.Binary
 
 
-class Error(Exception):
-    """Exception that is the base class of all other error exceptions.
-    You can use this to catch all errors with one single except statement.
-    """
-
-    pass
-
-
-class ParamEscaper(object):
+class ParamEscaper:
     def escape_args(self, parameters):
         if isinstance(parameters, dict):
             return {k: self.escape_item(v) for k, v in parameters.items()}
@@ -45,7 +38,7 @@ def escape_string(self, item):
         if isinstance(item, bytes):
             item = item.decode("utf-8")
         return "'{}'".format(
-            item.replace("\\", "\\\\").replace("'", "\\'").replace("$", "$$")
+            item.replace("\\", "\\\\").replace("'", "\\'").replace("$", "$$").replace("%", "%%")
         )
 
     def escape_item(self, item):
@@ -53,7 +46,9 @@ def escape_item(self, item):
             return "NULL"
         elif isinstance(item, (int, float)):
             return self.escape_number(item)
-        elif isinstance(item, datetime):
+        elif isinstance(item, decimal.Decimal):
+            return self.escape_number(item)
+        elif isinstance(item, (datetime, date)):
             return self.escape_string(item.strftime("%Y-%m-%d %H:%M:%S"))
         else:
             return self.escape_string(item)
@@ -115,7 +110,7 @@ def rollback(self):
         raise NotSupportedError("Transactions are not supported")  # pragma: no cover
 
 
-class Cursor(object):
+class Cursor:
     """These objects represent a database cursor, which is used to manage the context of a fetch
     operation.
 
@@ -173,22 +168,39 @@ def close(self):
     def execute(self, operation, parameters=None, is_response=True):
         """Prepare and execute a database operation (query or command)."""
 
+        #ToDo - Fix this, which is preventing the execution of blank DDL sunch as CREATE INDEX statements which aren't suppoorted
+        # Seems hard to fix when statements are coming from metadata.create_all()
+        if operation == "":
+            return
+
         self._reset_state()
 
         self._state = self._STATE_RUNNING
         self._uuid = uuid.uuid1()
 
-        if is_response:
-            rows = self._db.query_iter(operation)
-            schema = rows.schema()
-            columns = []
-            for field in schema.fields():
-                columns.append((field.name, field.data_type))
-            if self._state != self._STATE_RUNNING:
-                raise Exception("Should be running if processing response")
-            self._rows = rows
-            self._columns = columns
-            self._state = self._STATE_SUCCEEDED
+        try:
+            # operation = operation.replace('%%', '%')
+            if parameters:
+                query = operation % _escaper.escape_args(parameters)
+            else:
+                query = operation
+            query = query.replace('%%', '%')
+            if is_response:
+                rows = self._db.query_iter(query)
+                schema = rows.schema()
+                columns = []
+                for field in schema.fields():
+                    columns.append((field.name, field.data_type))
+                if self._state != self._STATE_RUNNING:
+                    raise Exception("Should be running if processing response")
+                self._rows = rows
+                self._columns = columns
+                self._state = self._STATE_SUCCEEDED
+            else:
+                self._db.exec(query)
+        except Exception as e:
+            # We have to raise dbAPI error
+            raise Error(str(e)) from e
 
     def executemany(self, operation, seq_of_parameters):
         """Prepare a database operation (query or command) and then execute it against all parameter
@@ -208,23 +220,28 @@ def executemany(self, operation, seq_of_parameters):
 
         m = RE_INSERT_VALUES.match(operation)
         if m:
-            q_prefix = m.group(1) % ()
-            q_values = m.group(2).rstrip()
-
-            for parameters in seq_of_parameters[:-1]:
-                values_list.append(q_values % _escaper.escape_args(parameters))
-            query = "{} {};".format(q_prefix, ",".join(values_list))
-            return self._db.raw(query)
-        for parameters in seq_of_parameters[:-1]:
-            self.execute(operation, parameters, is_response=False)
+            try:
+                q_prefix = m.group(1) #.replace('%%', '%') % ()
+                q_values = m.group(2).rstrip()
+
+                for parameters in seq_of_parameters:
+                    values_list.append(q_values % _escaper.escape_args(parameters))
+                query = "{} {};".format(q_prefix, ",".join(values_list))
+                self._db.exec(query)
+            except Exception as e:
+                # We have to raise dbAPI error
+                raise Error(str(e)) from e
+        else:
+            for parameters in seq_of_parameters:
+                self.execute(operation, parameters, is_response=False)
 
     def fetchone(self):
         """Fetch the next row of a query result set, returning a single sequence, or ``None`` when
         no more data is available."""
         if self._state == self._STATE_NONE:
-            raise Exception("No query yet")
+            raise Error("No query yet")
         if not self._rows:
-            raise Exception("No rows yet")
+            raise Error("No rows yet")
         else:
             self._rownumber += 1
             try:
@@ -243,7 +260,7 @@ def fetchmany(self, size=None):
         specified number of rows not being available, fewer rows may be returned.
         """
         if self._state == self._STATE_NONE:
-            raise Exception("No query yet")
+            raise Error("No query yet")
 
         if size is None:
             size = 1
@@ -262,7 +279,7 @@ def fetchall(self):
         (e.g. a list of tuples).
         """
         if self._state == self._STATE_NONE:
-            raise Exception("No query yet")
+            raise Error("No query yet")
 
         data = []
         if self._rows:
diff --git a/databend_sqlalchemy/databend_dialect.py b/databend_sqlalchemy/databend_dialect.py
index 8bc140d..641c801 100644
--- a/databend_sqlalchemy/databend_dialect.py
+++ b/databend_sqlalchemy/databend_dialect.py
@@ -2,40 +2,31 @@
 #
 # Note: parts of the file come from https://github.com/snowflakedb/snowflake-sqlalchemy
 #       licensed under the same Apache 2.0 License
-
+import decimal
 import re
-import sqlalchemy.types
-import sqlalchemy.util
+import datetime
 import sqlalchemy.types as sqltypes
-from types import ModuleType
-from typing import Any, Dict, List, Optional, Tuple, Union
-from sqlalchemy import exc as sa_exc
+from typing import Any, Dict, Optional, Union
 from sqlalchemy import util as sa_util
-from sqlalchemy.engine import default, reflection
-from sqlalchemy.sql import compiler, expression, text
-from sqlalchemy.sql.elements import quoted_name
+from sqlalchemy.engine import reflection
+from sqlalchemy.sql import compiler, text, bindparam
 from sqlalchemy.dialects.postgresql.base import PGCompiler, PGIdentifierPreparer
 from sqlalchemy.types import (
-    CHAR,
-    DATE,
-    DATETIME,
+    BIGINT,
     INTEGER,
     SMALLINT,
-    BIGINT,
     DECIMAL,
-    TIME,
-    TIMESTAMP,
+    NUMERIC,
     VARCHAR,
     BINARY,
     BOOLEAN,
     FLOAT,
-    REAL,
     JSON,
+    CHAR,
+    TIMESTAMP,
 )
 from sqlalchemy.engine import ExecutionContext, default
-
-# Column spec
-colspecs = {}
+from sqlalchemy.exc import DBAPIError, NoSuchTableError
 
 
 # Type decorators
@@ -52,20 +43,83 @@ def __init__(self, key_type, value_type):
         super(MAP, self).__init__()
 
 
+class DatabendDate(sqltypes.DATE):
+    __visit_name__ = "DATE"
+
+    _reg = re.compile(r"(\d+)-(\d+)-(\d+)")
+
+    def result_processor(self, dialect, coltype):
+        def process(value):
+            if isinstance(value, str):
+                m = self._reg.match(value)
+                if not m:
+                    raise ValueError(
+                        "could not parse %r as a date value" % (value,)
+                    )
+                return datetime.date(*[int(x or 0) for x in m.groups()])
+            else:
+                return value
+
+        return process
+
+
+class DatabendDateTime(sqltypes.DATETIME):
+    __visit_name__ = "DATETIME"
+
+    _reg = re.compile(r"(\d+)-(\d+)-(\d+) (\d+):(\d+):(\d+)")
+
+    def result_processor(self, dialect, coltype):
+        def process(value):
+            if isinstance(value, str):
+                m = self._reg.match(value)
+                if not m:
+                    raise ValueError(
+                        "could not parse %r as a datetime value" % (value,)
+                    )
+                return datetime.datetime(*[int(x or 0) for x in m.groups()])
+            else:
+                return value
+
+        return process
+
+
+class DatabendNumeric(sqltypes.Numeric):
+    def result_processor(self, dialect, type_):
+
+        orig = super().result_processor(dialect, type_)
+
+        def process(value):
+            if value is not None:
+                if self.decimal_return_scale:
+                    value = decimal.Decimal(f'{value:.{self.decimal_return_scale}f}')
+                else:
+                    value = decimal.Decimal(value)
+            if orig:
+                return orig(value)
+            return value
+
+        return process
+
+
 # Type converters
 ischema_names = {
+    "bigint": BIGINT,
     "int": INTEGER,
-    "int64": INTEGER,
+    "smallint": SMALLINT,
+    "tinyint": SMALLINT,
+    "int64": BIGINT,
     "int32": INTEGER,
-    "int16": INTEGER,
-    "int8": INTEGER,
-    "uint64": INTEGER,
+    "int16": SMALLINT,
+    "int8": SMALLINT,
+    "uint64": BIGINT,
     "uint32": INTEGER,
-    "uint16": INTEGER,
-    "uint8": INTEGER,
+    "uint16": SMALLINT,
+    "uint8": SMALLINT,
+    "numeric": NUMERIC,
     "decimal": DECIMAL,
-    "date": DATE,
-    "timestamp": DATETIME,
+    "date": DatabendDate,
+    "datetime": DatabendDateTime,
+    "timestamp": DatabendDateTime,
     "float": FLOAT,
     "double": FLOAT,
     "float64": FLOAT,
@@ -79,19 +133,24 @@ def __init__(self, key_type, value_type):
     "binary": BINARY,
 }
 
+# Column spec
+colspecs = {
+    sqltypes.Date: DatabendDate,
+    sqltypes.DateTime: DatabendDateTime,
+    sqltypes.DECIMAL: DatabendNumeric,
+    sqltypes.Numeric: DatabendNumeric,
+}
 
-class DatabendIdentifierPreparer(PGIdentifierPreparer):
-    def quote_identifier(self, value):
-        """Never quote identifiers."""
-        return self._escape_identifier(value)
 
-    def quote(self, ident, force=None):
-        if self._requires_quotes(ident):
-            return '"{}"'.format(ident)
-        return ident
+class DatabendIdentifierPreparer(PGIdentifierPreparer):
+    pass
 
 
 class DatabendCompiler(PGCompiler):
+    def get_select_precolumns(self, select, **kw):
+        # call the base implementation because Databend doesn't support DISTINCT ON
+        return super(PGCompiler, self).get_select_precolumns(select, **kw)
+
     def visit_count_func(self, fn, **kw):
         return "count{0}".format(self.process(fn.clause_expr, **kw))
 
@@ -104,12 +163,6 @@ def visit_now_func(self, fn, **kw):
     def visit_current_date_func(self, fn, **kw):
         return "today()"
 
-    def visit_true(self, element, **kw):
-        return "1"
-
-    def visit_false(self, element, **kw):
-        return "0"
-
     def visit_cast(self, cast, **kwargs):
         if self.dialect.supports_cast:
             return super(DatabendCompiler, self).visit_cast(cast, **kwargs)
@@ -131,28 +184,6 @@ def visit_concat_op_binary(self, binary, operator, **kw):
             self.process(binary.right),
         )
 
-    def visit_in_op_binary(self, binary, operator, **kw):
-        kw["literal_binds"] = True
-        return "%s IN %s" % (
-            self.process(binary.left, **kw),
-            self.process(binary.right, **kw),
-        )
-
-    def visit_notin_op_binary(self, binary, operator, **kw):
-        kw["literal_binds"] = True
-        return "%s NOT IN %s" % (
-            self.process(binary.left, **kw),
-            self.process(binary.right, **kw),
-        )
-
-    def visit_column(
-            self, column, add_to_result_map=None, include_table=True, **kwargs
-    ):
-        # Columns prefixed with table name are not supported
-        return super(DatabendCompiler, self).visit_column(
-            column, add_to_result_map=add_to_result_map, include_table=False, **kwargs
-        )
-
     def render_literal_value(self, value, type_):
         value = super(DatabendCompiler, self).render_literal_value(value, type_)
         if isinstance(type_, sqltypes.DateTime):
@@ -164,32 +195,89 @@ def render_literal_value(self, value, type_):
     def limit_clause(self, select, **kw):
         text = ""
         if select._limit_clause is not None:
-            text += "\n LIMIT " + self.process(select._limit_clause, **kw)
+            text += " \n LIMIT " + self.process(select._limit_clause, **kw)
         if select._offset_clause is not None:
-            text = "\n LIMIT "
             if select._limit_clause is None:
-                text += self.process(sql.literal(-1))
-            else:
-                text += "0"
-            text += "," + self.process(select._offset_clause, **kw)
+                text += "\n"
+            text += " OFFSET " + self.process(select._offset_clause, **kw)
         return text
 
     def for_update_clause(self, select, **kw):
         return ""  # Not supported
 
+    def visit_like_op_binary(self, binary, operator, **kw):
+        # escape = binary.modifiers.get("escape", None)
+        return "%s LIKE %s" % (
+            binary.left._compiler_dispatch(self, **kw),
+            binary.right._compiler_dispatch(self, **kw),
+        # ToDo - escape not yet supported
+        # ) + (
+        #     " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
+        #     if escape
+        #     else ""
+        )
+
+    def visit_not_like_op_binary(self, binary, operator, **kw):
+        # escape = binary.modifiers.get("escape", None)
+        return "%s NOT LIKE %s" % (
+            binary.left._compiler_dispatch(self, **kw),
+            binary.right._compiler_dispatch(self, **kw),
+        # ToDo - escape not yet supported
+        # ) + (
+        #     " ESCAPE " + self.render_literal_value(escape, sqltypes.STRINGTYPE)
+        #     if escape
+        #     else ""
+        )
+
 
 class DatabendExecutionContext(default.DefaultExecutionContext):
     @sa_util.memoized_property
     def should_autocommit(self):
         return False  # No DML supported, never autocommit
 
+    def create_server_side_cursor(self):
+        return self._dbapi_connection.cursor()
+
+    def create_default_cursor(self):
+        return self._dbapi_connection.cursor()
+
 
 class DatabendTypeCompiler(compiler.GenericTypeCompiler):
-    def visit_ARRAY(self, type, **kw):
-        return "Array(%s)" % type
+    def visit_ARRAY(self, type_, **kw):
+        return "Array(%s)" % type_
+
+    def Visit_MAP(self, type_, **kw):
+        return "Map(%s)" % type_
+
+    def visit_NUMERIC(self, type_, **kw):
+        if type_.precision is None:
+            return self.visit_DECIMAL(sqltypes.DECIMAL(38, 10), **kw)
+        if type_.scale is None:
+            return self.visit_DECIMAL(sqltypes.DECIMAL(38, 10), **kw)
+        return self.visit_DECIMAL(type_, **kw)
 
-    def Visit_MAP(self, type, **kw):
-        return "Map(%s)" % type
+    def visit_NVARCHAR(self, type_, **kw):
+        return self.visit_VARCHAR(type_, **kw)
+
+
+class DatabendDDLCompiler(compiler.DDLCompiler):
+
+    def visit_primary_key_constraint(self, constraint, **kw):
+        return ""
+
+    def visit_foreign_key_constraint(self, constraint, **kw):
+        return ""
+
+    def create_table_constraints(
+        self, table, _include_foreign_key_constraints=None, **kw
+    ):
+        return ""
+
+    def visit_create_index(self, create, include_schema=False, include_table_schema=True, **kw):
+        return ""
+
+    def visit_drop_index(self, drop, **kw):
+        return ""
 
 
 class DatabendDialect(default.DefaultDialect):
@@ -201,7 +289,14 @@ class DatabendDialect(default.DefaultDialect):
     supports_sane_rowcount = False
     supports_sane_multi_rowcount = False
     supports_native_boolean = True
+    supports_native_decimal = True
     supports_alter = True
+    supports_comments = False
+    supports_empty_insert = False
+    supports_is_distinct_from = False
+
+    supports_statement_cache = False
+    supports_server_side_cursors = True
 
     max_identifier_length = 127
     default_paramstyle = "pyformat"
@@ -209,12 +304,15 @@ class DatabendDialect(default.DefaultDialect):
     ischema_names = ischema_names
     convert_unicode = True
     returns_unicode_strings = True
+    returns_native_bytes = True
+    div_is_floordiv = False
     description_encoding = None
     postfetch_lastrowid = False
 
     preparer = DatabendIdentifierPreparer
     type_compiler = DatabendTypeCompiler
     statement_compiler = DatabendCompiler
+    ddl_compiler = DatabendDDLCompiler
     execution_ctx_cls = DatabendExecutionContext
 
     # Required for PG-based compiler
@@ -228,14 +326,24 @@ def __init__(
 
     @classmethod
     def dbapi(cls):
+        return cls.import_dbapi()
+
+    @classmethod
+    def import_dbapi(cls):
         try:
             import databend_sqlalchemy.connector as connector
         except Exception:
             import connector
         return connector
 
-    def initialize(self, connection):
-        pass
+    def _get_server_version_info(self, connection):
+        val = connection.scalar(text("SELECT VERSION()"))
+        m = re.match(r"(?:.*)v(\d+).(\d+).(\d+)-([^\(]+)(?:\()", val)
+        if not m:
+            raise AssertionError(
+                "Could not determine version from string '%s'" % val
+            )
+        return tuple(int(x) for x in m.group(1, 2, 3) if x is not None)
 
     def connect(self, *cargs, **cparams):
         # inherits the docstring from interfaces.Dialect.connect
@@ -252,57 +360,106 @@ def create_connect_args(self, url):
 
         return ([], kwargs)
 
+    def create_server_side_cursor(self):
+        return self.create_default_cursor()
+
     def _get_default_schema_name(self, connection):
-        return connection.scalar(text("select currentDatabase()"))
+        return connection.scalar(text("SELECT currentDatabase()"))
 
+    @reflection.cache
     def get_schema_names(self, connection, **kw):
-        return [row[0] for row in connection.execute("SHOW DATABASES")]
-
-    def get_view_names(self, connection, schema=None, **kw):
-        return self.get_table_names(connection, schema, **kw)
+        return [row[0] for row in connection.execute(text("SHOW DATABASES"))]
 
     def _get_table_columns(self, connection, table_name, schema):
-        full_table = table_name
-        if schema:
-            full_table = schema + "." + table_name
-        # This needs the table name to be unescaped (no backticks).
-        return connection.execute(text("DESC {}".format(full_table))).fetchall()
+        if schema is None:
+            schema = self.default_schema_name
+        quote_table_name = self.identifier_preparer.quote_identifier(table_name)
+        quote_schema = self.identifier_preparer.quote_identifier(schema)
+
+        return connection.execute(text(f"DESC {quote_schema}.{quote_table_name}")).fetchall()
 
+    @reflection.cache
     def has_table(self, connection, table_name, schema=None, **kw):
-        full_table = table_name
-        if schema:
-            full_table = schema + "." + table_name
-        for r in connection.execute(text("EXISTS TABLE {}".format(full_table))):
-            if r[0] == 1:
-                return True
+        if schema is None:
+            schema = self.default_schema_name
+        quote_table_name = self.identifier_preparer.quote_identifier(table_name)
+        quote_schema = self.identifier_preparer.quote_identifier(schema)
+        query = f"""EXISTS TABLE {quote_schema}.{quote_table_name}"""
+        r = connection.scalar(text(query))
+        if r == 1:
+            return True
         return False
 
     @reflection.cache
     def get_columns(self, connection, table_name, schema=None, **kw):
-        query = """
-            select column_name, data_type, is_nullable
+        query = text(
+            """
+            select column_name, column_type, is_nullable
             from information_schema.columns
-            where table_name = '{table_name}'
-        """.format(
-            table_name=table_name
+            where table_name = :table_name
+            and table_schema = :schema_name
+            """
+        ).bindparams(
+            bindparam("table_name", type_=sqltypes.UnicodeText),
+            bindparam("schema_name", type_=sqltypes.Unicode)
         )
+        if schema is None:
+            schema = self.default_schema_name
+        result = connection.execute(query, dict(table_name=table_name, schema_name=schema))
 
-        if schema:
-            query = "{query} and table_schema = '{schema}'".format(
-                query=query, schema=schema
-            )
-
-        result = connection.execute(text(query))
-
-        return [
+        cols = [
             {
                 "name": row[0],
-                "type": ischema_names[extract_nullable_string(row[1]).lower()],
+                "type": self._get_column_type(row[1]),
                 "nullable": get_is_nullable(row[2]),
                 "default": None,
             }
             for row in result
         ]
+        if not cols and not self.has_table(connection, table_name, schema):
+            raise NoSuchTableError(table_name)
+        return cols
+
+    @reflection.cache
+    def get_view_definition(self, connection, view_name, schema=None, **kw):
+        if schema is None:
+            schema = self.default_schema_name
+        quote_schema = self.identifier_preparer.quote_identifier(schema)
+        quote_view_name = self.identifier_preparer.quote_identifier(view_name)
+        full_view_name = f"{quote_schema}.{quote_view_name}"
+
+        # ToDo : perhaps can be removed if we get `SHOW CREATE VIEW`
+        if view_name not in self.get_view_names(connection, schema):
+            raise NoSuchTableError(full_view_name)
+
+        query = f"""SHOW CREATE TABLE {full_view_name}"""
+        try:
+            view_def = connection.execute(text(query)).first()
+            return view_def[1]
+        except DBAPIError as e:
+            if '1025' in e.orig.message:  #ToDo: The errors need parsing properly
+                raise NoSuchTableError(full_view_name) from e
+
+    def _get_column_type(self, column_type):
+        pattern = r'(?:Nullable)*(?:\()*(\w+)(?:\((.*?)\))?(?:\))*'
+        match = re.match(pattern, column_type)
+        if match:
+            type_str = match.group(1).lower()
+            charlen = match.group(2)
+            args = ()
+            kwargs = {}
+            if type_str == 'decimal':
+                if charlen:
+                    # e.g.'18, 5'
+                    prec, scale = charlen.split(", ")
+                    args = (int(prec), int(scale))
+            elif charlen:
+                args = (int(charlen),)
+
+            coltype = self.ischema_names[type_str]
+            return coltype(*args, **kwargs)
+        else:
+            return None
 
     @reflection.cache
     def get_foreign_keys(self, connection, table_name, schema=None, **kw):
@@ -318,15 +475,39 @@ def get_pk_constraint(self, connection, table_name, schema=None, **kw):
     def get_indexes(self, connection, table_name, schema=None, **kw):
         return []
 
-    # @reflection.cache
+    @reflection.cache
     def get_table_names(self, connection, schema=None, **kw):
-        query = "select table_name from information_schema.tables"
-        if schema:
-            query = "{query} where table_schema = '{schema}'".format(
-                query=query, schema=schema
-            )
+        query = text("""
+            select table_name
+            from information_schema.tables
+            where table_schema = :schema_name
+            and engine NOT LIKE '%VIEW%'
+            """
+        ).bindparams(
+            bindparam("schema_name", type_=sqltypes.Unicode)
+        )
+        if schema is None:
+            schema = self.default_schema_name
+
+        result = connection.execute(query, dict(schema_name=schema))
+        return [row[0] for row in result]
+
+    @reflection.cache
+    def get_view_names(self, connection, schema=None, **kw):
+        query = text(
+            """
+            select table_name
+            from information_schema.tables
+            where table_schema = :schema_name
+            and engine LIKE '%VIEW%'
+            """
+        ).bindparams(
+            bindparam("schema_name", type_=sqltypes.Unicode)
+        )
+        if schema is None:
+            schema = self.default_schema_name
 
-        result = connection.execute(text(query))
+        result = connection.execute(query, dict(schema_name=schema))
         return [row[0] for row in result]
 
     def do_rollback(self, dbapi_connection):
diff --git a/databend_sqlalchemy/provision.py b/databend_sqlalchemy/provision.py
new file mode 100644
index 0000000..68586f8
--- /dev/null
+++ b/databend_sqlalchemy/provision.py
@@ -0,0 +1,38 @@
+
+from sqlalchemy.testing.provision import create_db
+from sqlalchemy.testing.provision import drop_db
+from sqlalchemy.testing.provision import configure_follower
+
+
+@create_db.for_db("databend")
+def _databend_create_db(cfg, eng, ident):
+    with eng.begin() as conn:
+        try:
+            _databend_drop_db(cfg, conn, ident)
+        except Exception:
+            pass
+
+    with eng.begin() as conn:
+        conn.exec_driver_sql(
+            "CREATE DATABASE IF NOT EXISTS %s " % ident
+        )
+        conn.exec_driver_sql(
+            "CREATE DATABASE IF NOT EXISTS %s_test_schema" % ident
+        )
+        conn.exec_driver_sql(
+            "CREATE DATABASE IF NOT EXISTS %s_test_schema_2" % ident
+        )
+
+
+@drop_db.for_db("databend")
+def _databend_drop_db(cfg, eng, ident):
+    with eng.begin() as conn:
+        conn.exec_driver_sql("DROP DATABASE IF EXISTS %s_test_schema" % ident)
+        conn.exec_driver_sql("DROP DATABASE IF EXISTS %s_test_schema_2" % ident)
+        conn.exec_driver_sql("DROP DATABASE IF EXISTS %s" % ident)
+
+
+@configure_follower.for_db("databend")
+def _databend_configure_follower(config, ident):
+    config.test_schema = "%s_test_schema" % ident
+    config.test_schema_2 = "%s_test_schema_2" % ident
diff --git a/databend_sqlalchemy/requirements.py b/databend_sqlalchemy/requirements.py
new file mode 100644
index 0000000..560709b
--- /dev/null
+++ b/databend_sqlalchemy/requirements.py
@@ -0,0 +1,188 @@
+
+from sqlalchemy.testing.requirements import SuiteRequirements
+
+from sqlalchemy.testing import exclusions
+
+
+class Requirements(SuiteRequirements):
+
+    @property
+    def foreign_keys(self):
+        """Target database must support foreign keys."""
+
+        return exclusions.closed()  # Currently no foreign keys in Databend
+
+    @property
+    def binary_comparisons(self):
+        """target database/driver can allow BLOB/BINARY fields to be compared
+        against a bound parameter value.
+        """
+        return exclusions.closed()  # Currently no binary type in Databend
+
+    @property
+    def binary_literals(self):
+        """target backend supports simple binary literals, e.g. an
+        expression like::
+
+            SELECT CAST('foo' AS BINARY)
+
+        Where ``BINARY`` is the type emitted from :class:`.LargeBinary`,
+        e.g. it could be ``BLOB`` or similar.
+
+        Basically fails on Oracle.
+
+        """
+        return exclusions.closed()  # Currently no binary type in Databend
+
+    @property
+    def temporary_tables(self):
+        """target database supports temporary tables"""
+        return exclusions.closed()  # Databend does not currently support temporary tables
+
+    @property
+    def temp_table_reflection(self):
+        return exclusions.closed()
+
+    @property
+    def self_referential_foreign_keys(self):
+        """Target database must support self-referential foreign keys."""
+
+        return exclusions.closed()  # Databend does not currently support foreign keys
+
+    @property
+    def foreign_key_ddl(self):
+        """Target database must support the DDL phrases for FOREIGN KEY."""
+
+        return exclusions.closed()  # Databend does not currently support foreign keys
+
+    @property
+    def index_reflection(self):
+        return exclusions.closed()   # Databend does not currently support indexes
+
+    @property
+    def primary_key_constraint_reflection(self):
+        return exclusions.closed()  # Databend does not currently support primary keys
+
+    @property
+    def foreign_key_constraint_reflection(self):
+        return exclusions.closed()  # Databend does not currently support foreign keys
+
+    @property
+    def unique_constraint_reflection(self):
+        """target dialect supports reflection of unique constraints"""
+        return exclusions.closed()  # Databend does not currently support unique constraints
+
+    @property
+    def duplicate_key_raises_integrity_error(self):
+        """target dialect raises IntegrityError when reporting an INSERT
+        with a primary key violation.  (hint: it should)
+
+        """
+        return exclusions.closed()   # Databend does not currently support primary keys
+
+    @property
+    def sql_expression_limit_offset(self):
+        """target database can render LIMIT and/or OFFSET with a complete
+        SQL expression, such as one that uses the addition operator.
+        parameter
+        """
+
+        return exclusions.closed()  # Databend does not currently support expressions in limit/offset
+
+    @property
+    def autoincrement_without_sequence(self):
+        """If autoincrement=True on a column does not require an explicit
+        sequence. This should be false only for oracle.
+        """
+        return exclusions.closed()
+
+    @property
+    def datetime_timezone(self):
+        """target dialect supports representation of Python
+        datetime.datetime() with tzinfo with DateTime(timezone=True)."""
+
+        return exclusions.closed()
+
+    @property
+    def time_timezone(self):
+        """target dialect supports representation of Python
+        datetime.time() with tzinfo with Time(timezone=True)."""
+
+        return exclusions.closed()
+
+    # @property
+    # def datetime_implicit_bound(self):
+    #     """target dialect when given a datetime object will bind it such
+    #     that the database server knows the object is a datetime, and not
+    #     a plain string.
+    #
+    #     """
+    #     return exclusions.closed()  #  `SELECT '2012-10-15 12:57:18' AS thing` does not yield a timestamp in Databend
+
+    @property
+    def datetime_microseconds(self):
+        """target dialect supports representation of Python
+        datetime.datetime() with microsecond objects."""
+
+        return exclusions.closed()
+
+    @property
+    def timestamp_microseconds(self):
+        """target dialect supports representation of Python
+        datetime.datetime() with microsecond objects but only
+        if TIMESTAMP is used."""
+        return exclusions.closed()
+
+    @property
+    def time(self):
+        """target dialect supports representation of Python
+        datetime.time() objects."""
+
+        return exclusions.closed()
+
+    @property
+    def time_microseconds(self):
+        """target dialect supports representation of Python
+        datetime.time() with microsecond objects."""
+
+        return exclusions.closed()
+
+    @property
+    def autoincrement_insert(self):
+        """target platform generates new surrogate integer primary key values
+        when insert() is executed, excluding the pk column."""
+
+        return exclusions.closed()
+
+    @property
+    def views(self):
+        """Target database must support VIEWs."""
+
+        return exclusions.open()
+
+    @property
+    def unicode_data(self):
+        """Target database/dialect must support Python unicode objects with
+        non-ASCII characters represented, delivered as bound parameters
+        as well as in result rows.
+
+        """
+        return exclusions.closed()
+
+    @property
+    def precision_generic_float_type(self):
+        """target backend will return native floating point numbers with at
+        least seven decimal places when using the generic Float type.
+
+        """
+        return exclusions.closed()  # I couldn't get the test for this one working, maybe when the driver returns a python decimal?
+
+    @property
+    def array_type(self):
+        return exclusions.closed()
+
+    @property
+    def float_is_numeric(self):
+        """target backend uses Numeric for Float/Dual"""
+
+        return exclusions.closed()
diff --git a/setup.cfg b/setup.cfg
index b0133fd..7adf632 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -27,8 +27,8 @@ project_urls =
 [options]
 packages = find:
 install_requires =
-    databend-driver>=0.13.0
-    sqlalchemy>1.3.21
+    databend-driver>=0.16.0
+    sqlalchemy>=1.4
 python_requires = >=3.7
 package_dir =
     = .
@@ -39,6 +39,7 @@ where = .
 [options.entry_points]
 sqlalchemy.dialects =
     databend = databend_sqlalchemy.databend_dialect:DatabendDialect
+    databend.databend = databend_sqlalchemy.databend_dialect:DatabendDialect
 superset.db_engine_specs =
     databend = databend_sqlalchemy.cc_superset.engine:DatabendEngineSpec
 
@@ -47,8 +48,9 @@ dev =
     devtools==0.7.0
     mock==4.0.3
     pre-commit==2.15.0
-    pytest==6.2.5
+    pytest==8.1.1
     pytest-cov==3.0.0
+    pytest-xdist==3.5.0
     sqlalchemy-stubs==0.4
 
 superset = apache_superset>=1.4.1
@@ -59,3 +61,16 @@ per-file-ignores = __init__.py:F401
 ignore = E203, W503
 ban-relative-imports = True
 inline-quotes = "
+
+[tool:pytest]
+addopts= --tb native -n4 -v -r fxX --maxfail=25 -p no:warnings
+python_files=tests/*test_*.py
+# python_functions=test_round_trip_same_named_column
+# log_level=DEBUG
+
+[sqla_testing]
+requirement_cls=databend_sqlalchemy.requirements:Requirements
+profile_file = .profiles.txt
+
+[db]
+default=databend://databend:databend@localhost:8000/default?sslmode=disable


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 0fc50f7..b448bbe 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -1,4 +1,4 @@
-name: CI
+name: CI TEST
 
 on:
   push:
@@ -24,7 +24,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Setup Python
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v5
         with:
           python-version: '3.11'
 
@@ -39,14 +39,8 @@ jobs:
           docker logs ${cid}
           curl -v http://localhost:8000/v1/health
 
-      - name: Unit Test
+      - name: SQLAlchemy Test Suite
         env:
           TEST_DATABEND_DSN: "databend://databend:databend@localhost:8000/default?sslmode=disable"
         run: |
-          pipenv run pytest -s tests/unit
-
-      - name: Integration Test
-        env:
-          TEST_DATABEND_DSN: "databend://databend:databend@localhost:8000/default?sslmode=disable"
-        run: |
-          pipenv run pytest -s tests/integration
+          pipenv run pytest -s .
diff --git a/databend_sqlalchemy/test.py b/databend_sqlalchemy/test.py
deleted file mode 100644
index c613fb9..0000000
--- a/databend_sqlalchemy/test.py
+++ /dev/null
@@ -1,20 +0,0 @@
-from databend_sqlalchemy import connector
-
-
-def test():
-    conn = connector.connect(
-        "databend://databend:databend@localhost:8000/default?sslmode=disable"
-    )
-    cursor = conn.cursor()
-    cursor.execute(
-        "select null as db, name as name, database as schema, if(engine = 'VIEW', 'view', 'table') "
-        "as type from system.tables where database = 'default';"
-    )
-    # print(cursor.fetchone())
-    print(cursor.fetchall())
-    print(cursor.description)
-    conn.close()
-
-
-if __name__ == "__main__":
-    test()
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..d5117f1
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,9 @@
+from sqlalchemy.dialects import registry
+import pytest
+
+registry.register("databend.databend", "databend_sqlalchemy.databend_dialect", "DatabendDialect")
+registry.register("databend", "databend_sqlalchemy.databend_dialect", "DatabendDialect")
+
+pytest.register_assert_rewrite("sa.testing.assertions")
+
+from sqlalchemy.testing.plugin.pytestplugin import *
diff --git a/tests/test_sqlalchemy.py b/tests/test_sqlalchemy.py
new file mode 100644
index 0000000..5fa01a1
--- /dev/null
+++ b/tests/test_sqlalchemy.py
@@ -0,0 +1,279 @@
+# tests/test_suite.py
+
+from sqlalchemy.testing.suite import *
+
+from sqlalchemy.testing.suite import ComponentReflectionTestExtra as _ComponentReflectionTestExtra
+from sqlalchemy.testing.suite import ComponentReflectionTest as _ComponentReflectionTest
+from sqlalchemy.testing.suite import DeprecatedCompoundSelectTest as _DeprecatedCompoundSelectTest
+from sqlalchemy.testing.suite import BooleanTest as _BooleanTest
+from sqlalchemy.testing.suite import BinaryTest as _BinaryTest
+from sqlalchemy.testing.suite import CompoundSelectTest as _CompoundSelectTest
+from sqlalchemy.testing.suite import HasIndexTest as _HasIndexTest
+from sqlalchemy.testing.suite import InsertBehaviorTest as _InsertBehaviorTest
+from sqlalchemy.testing.suite import LikeFunctionsTest as _LikeFunctionsTest
+from sqlalchemy.testing.suite import LongNameBlowoutTest as _LongNameBlowoutTest
+from sqlalchemy.testing.suite import QuotedNameArgumentTest as _QuotedNameArgumentTest
+from sqlalchemy.testing.suite import JoinTest as _JoinTest
+from sqlalchemy.testing.suite import BizarroCharacterFKResolutionTest as _BizarroCharacterFKResolutionTest
+from sqlalchemy.testing.suite import ServerSideCursorsTest as _ServerSideCursorsTest
+from sqlalchemy import types as sql_types
+from sqlalchemy import testing, select
+from sqlalchemy.testing import config, eq_
+
+
+class ComponentReflectionTest(_ComponentReflectionTest):
+    @testing.requires.index_reflection
+    def test_get_indexes(self):
+        pass
+
+
+class ComponentReflectionTestExtra(_ComponentReflectionTestExtra):
+
+    @testing.requires.table_reflection
+    def test_varchar_reflection(self, connection, metadata):
+        typ = self._type_round_trip(
+            connection, metadata, sql_types.String(52)
+        )[0]
+        assert isinstance(typ, sql_types.String)
+        # eq_(typ.length, 52)  #  No length in Databend
+
+
+class BooleanTest(_BooleanTest):
+    __backend__ = True
+
+    def test_whereclause(self):
+        """
+        This is overridden from Ancestor implementation because Databend does not support `WHERE NOT true|false`
+        Please compare this version with the overridden test
+        """
+        # testing "WHERE <column>" renders a compatible expression
+        boolean_table = self.tables.boolean_table
+
+        with config.db.begin() as conn:
+            conn.execute(
+                boolean_table.insert(),
+                [
+                    {"id": 1, "value": True, "unconstrained_value": True},
+                    {"id": 2, "value": False, "unconstrained_value": False},
+                ],
+            )
+
+            eq_(
+                conn.scalar(
+                    select(boolean_table.c.id).where(boolean_table.c.value)
+                ),
+                1,
+            )
+            eq_(
+                conn.scalar(
+                    select(boolean_table.c.id).where(
+                        boolean_table.c.unconstrained_value
+                    )
+                ),
+                1,
+            )
+
+
+class CompoundSelectTest(_CompoundSelectTest):
+
+    @testing.skip("databend")
+    def test_limit_offset_aliased_selectable_in_unions(self):
+        pass
+
+    @testing.skip("databend")
+    def test_limit_offset_selectable_in_unions(self):
+        pass
+
+    @testing.skip("databend")
+    def test_limit_offset_in_unions_from_alias(self):
+        pass
+
+
+class DeprecatedCompoundSelectTest(_DeprecatedCompoundSelectTest):
+    @testing.skip("databend")
+    def test_limit_offset_aliased_selectable_in_unions(self):
+        pass
+
+    @testing.skip("databend")
+    def test_limit_offset_selectable_in_unions(self):
+        pass
+
+    @testing.skip("databend")
+    def test_limit_offset_in_unions_from_alias(self):
+        pass
+
+
+class HasIndexTest(_HasIndexTest):
+    __requires__ = ('index_reflection',)
+
+
+class InsertBehaviorTest(_InsertBehaviorTest):
+    @testing.skip("databend")  # required autoinc columns
+    def test_insert_from_select_autoinc(self, connection):
+        pass
+
+    @testing.skip("databend")  # required autoinc columns
+    def test_insert_from_select_autoinc_no_rows(self, connection):
+        pass
+
+    @testing.skip("databend")  # required autoinc columns
+    def test_no_results_for_non_returning_insert(self, connection):
+        pass
+
+
+class LikeFunctionsTest(_LikeFunctionsTest):
+
+    @testing.skip("databend")
+    def test_contains_autoescape(self):
+        pass
+
+    @testing.skip("databend")
+    def test_contains_escape(self):
+        pass
+
+    @testing.skip("databend")
+    def test_contains_autoescape_escape(self):
+        pass
+
+    @testing.skip("databend")
+    def test_endswith_autoescape(self):
+        pass
+
+    @testing.skip("databend")
+    def test_endswith_escape(self):
+        pass
+
+    @testing.skip("databend")
+    def test_endswith_autoescape_escape(self):
+        pass
+
+    @testing.skip("databend")
+    def test_startswith_autoescape(self):
+        pass
+
+    @testing.skip("databend")
+    def test_startswith_escape(self):
+        pass
+
+    @testing.skip("databend")
+    def test_startswith_autoescape_escape(self):
+        pass
+
+
+class LongNameBlowoutTest(_LongNameBlowoutTest):
+    __requires__ = ("index_reflection",)  # This will do to make it skip for now
+
+
+class QuotedNameArgumentTest(_QuotedNameArgumentTest):
+    def quote_fixtures(fn):
+        return testing.combinations(
+            ("quote ' one",),
+            ('quote " two', testing.requires.symbol_names_w_double_quote),
+        )(fn)
+
+    @quote_fixtures
+    @testing.skip("databend")
+    def test_get_pk_constraint(self, name):
+        pass
+
+    @quote_fixtures
+    @testing.skip("databend")
+    def test_get_foreign_keys(self, name):
+        pass
+
+    @quote_fixtures
+    @testing.skip("databend")
+    def test_get_indexes(self, name):
+        pass
+
+
+class JoinTest(_JoinTest):
+    __requires__ = ("foreign_keys",)
+
+
+class BizarroCharacterFKResolutionTest(_BizarroCharacterFKResolutionTest):
+    __requires__ = ("foreign_keys",)
+
+
+class BinaryTest(_BinaryTest):
+
+    # ToDo - get this working, failing because cannot substitute bytes parameter into sql statement
+    # CREATE TABLE binary_test (x binary not null)
+    # INSERT INTO binary_test (x) values (b'7\xe7\x9f') ???
+    # It is possible to do this
+    # INSERT INTO binary_test (x) values (TO_BINARY('7\xe7\x9f'))
+    # but that's not really a solution I don't think
+    @testing.skip("databend")
+    def test_binary_roundtrip(self):
+        pass
+
+    @testing.skip("databend")
+    def test_pickle_roundtrip(self):
+        pass
+
+
+class ServerSideCursorsTest(_ServerSideCursorsTest):
+
+    def _is_server_side(self, cursor):
+        # ToDo - requires implementation of `stream_results` option, so True always for now
+        if self.engine.dialect.driver == "databend":
+            return True
+        return super()
+
+    # ToDo - The commented out testing combinations here should be reviewed when `stream_results` is implemented
+    @testing.combinations(
+        ("global_string", True, "select 1", True),
+        ("global_text", True, text("select 1"), True),
+        ("global_expr", True, select(1), True),
+        # ("global_off_explicit", False, text("select 1"), False),
+        (
+            "stmt_option",
+            False,
+            select(1).execution_options(stream_results=True),
+            True,
+        ),
+        # (
+        #     "stmt_option_disabled",
+        #     True,
+        #     select(1).execution_options(stream_results=False),
+        #     False,
+        # ),
+        ("for_update_expr", True, select(1).with_for_update(), True),
+        # TODO: need a real requirement for this, or dont use this test
+        # (
+        #     "for_update_string",
+        #     True,
+        #     "SELECT 1 FOR UPDATE",
+        #     True,
+        #     testing.skip_if(["sqlite", "mssql"]),
+        # ),
+        # ("text_no_ss", False, text("select 42"), False),
+        (
+            "text_ss_option",
+            False,
+            text("select 42").execution_options(stream_results=True),
+            True,
+        ),
+        id_="iaaa",
+        argnames="engine_ss_arg, statement, cursor_ss_status",
+    )
+    def test_ss_cursor_status(
+        self, engine_ss_arg, statement, cursor_ss_status
+    ):
+        super()
+
+    @testing.skip("databend")  # ToDo - requires implementation of `stream_results` option
+    def test_stmt_enabled_conn_option_disabled(self):
+        pass
+
+    @testing.skip("databend")  # ToDo - requires implementation of `stream_results` option
+    def test_aliases_and_ss(self):
+        pass
+
+    @testing.skip("databend")  # Skipped because requires auto increment primary key
+    def test_roundtrip_fetchall(self):
+        pass
+
+    @testing.skip("databend")  # Skipped because requires auto increment primary key
+    def test_roundtrip_fetchmany(self):
+        pass
diff --git a/tests/unit/test_databend_dialect.py b/tests/unit/test_databend_dialect.py
index 0a821f2..26b0fe9 100644
--- a/tests/unit/test_databend_dialect.py
+++ b/tests/unit/test_databend_dialect.py
@@ -13,6 +13,8 @@
     DatabendDialect,
     DatabendIdentifierPreparer,
     DatabendTypeCompiler,
+    INTEGER, BOOLEAN, BINARY,
+    DatabendDate
 )
 from databend_sqlalchemy.databend_dialect import dialect as dialect_definition
 
@@ -60,18 +62,43 @@ def test_table_names(
         assert result == ["table1", "table2"]
         connection.execute.assert_called_once()
         assert str(connection.execute.call_args[0][0].compile()) == str(
-            text("select table_name from information_schema.tables").compile()
+            text("""
+            select table_name
+            from information_schema.tables
+            where table_schema = :schema_name
+            and engine NOT LIKE '%VIEW%'
+            """).compile()
         )
+        assert connection.execute.call_args[0][1] == {'schema_name': None}
         connection.execute.reset_mock()
+        # Test default schema
+        dialect.default_schema_name = 'some-schema'
+        result = dialect.get_table_names(connection)
+        assert result == ["table1", "table2"]
+        connection.execute.assert_called_once()
+        assert str(connection.execute.call_args[0][0].compile()) == str(
+            text("""
+            select table_name
+            from information_schema.tables
+            where table_schema = :schema_name
+            and engine NOT LIKE '%VIEW%'
+            """).compile()
+        )
+        assert connection.execute.call_args[0][1] == {'schema_name': 'some-schema'}
+        connection.execute.reset_mock()
+        # Test specified schema
         result = dialect.get_table_names(connection, schema="schema")
         assert result == ["table1", "table2"]
         connection.execute.assert_called_once()
         assert str(connection.execute.call_args[0][0].compile()) == str(
-            text(
-                "select table_name from information_schema.tables"
-                " where table_schema = 'schema'"
-            ).compile()
+            text("""
+            select table_name
+            from information_schema.tables
+            where table_schema = :schema_name
+            and engine NOT LIKE '%VIEW%'
+            """).compile()
         )
+        assert connection.execute.call_args[0][1] == {'schema_name': 'schema'}
 
     def test_view_names(
             self, dialect: DatabendDialect, connection: mock.Mock(spec=MockDBApi)
@@ -103,42 +130,46 @@ def getitem(self, idx):
         ]
 
         expected_query = """
-            select column_name, data_type, is_nullable
+            select column_name, column_type, is_nullable
             from information_schema.columns
-            where table_name = 'table'
-        """
+            where table_name = :table_name
+            and table_schema = :schema_name
+            """
 
-        expected_query_schema = expected_query + " and table_schema = 'schema'"
 
-        for call, expected_query in (
-                (lambda: dialect.get_columns(connection, "table"), expected_query),
+        for call, expected_params in (
+                (
+                    lambda: dialect.get_columns(connection, "table"),
+                    {'table_name': 'table', 'schema_name': None},
+                ),
                 (
-                        lambda: dialect.get_columns(connection, "table", "schema"),
-                        expected_query_schema,
+                    lambda: dialect.get_columns(connection, "table", "schema"),
+                    {'table_name': 'table', 'schema_name': 'schema'},
                 ),
         ):
-            assert call() == [
+            result = call()
+            assert result == [
                 {
                     "name": "name1",
-                    "type": sqlalchemy.types.INTEGER,
+                    "type": INTEGER,
                     "nullable": True,
                     "default": None,
                 },
                 {
                     "name": "name2",
-                    "type": sqlalchemy.types.DATE,
+                    "type": DatabendDate,
                     "nullable": False,
                     "default": None,
                 },
                 {
                     "name": "name3",
-                    "type": sqlalchemy.types.BOOLEAN,
+                    "type": BOOLEAN,
                     "nullable": True,
                     "default": None,
                 },
                 {
                     "name": "name4",
-                    "type": sqlalchemy.types.BINARY,
+                    "type": BINARY,
                     "nullable": True,
                     "default": None,
                 },
@@ -147,6 +178,7 @@ def getitem(self, idx):
             assert str(connection.execute.call_args[0][0].compile()) == str(
                 text(expected_query).compile()
             )
+            assert connection.execute.call_args[0][1] == expected_params
             connection.execute.reset_mock()
 
 
@@ -157,10 +189,10 @@ def test_get_is_nullable():
 
 def test_types():
     assert databend_sqlalchemy.databend_dialect.CHAR is sqlalchemy.sql.sqltypes.CHAR
-    assert databend_sqlalchemy.databend_dialect.DATE is sqlalchemy.sql.sqltypes.DATE
-    assert (
-            databend_sqlalchemy.databend_dialect.DATETIME
-            is sqlalchemy.sql.sqltypes.DATETIME
+    assert issubclass(databend_sqlalchemy.databend_dialect.DatabendDate, sqlalchemy.sql.sqltypes.DATE)
+    assert issubclass(
+            databend_sqlalchemy.databend_dialect.DatabendDateTime,
+            sqlalchemy.sql.sqltypes.DATETIME,
     )
     assert (
             databend_sqlalchemy.databend_dialect.INTEGER is sqlalchemy.sql.sqltypes.INTEGER