diff --git a/README.md b/README.md
index 57c1a865..09a4c5e0 100644
--- a/README.md
+++ b/README.md
@@ -317,9 +317,39 @@ The transaction is created when the first SQL statement is executed.
 exits the *with* context and the queries succeed, otherwise
 `trino.dbapi.Connection.rollback()` will be called.
 
-## Development
+# Improved Python types
 
-### Getting Started With Development
+If you enable the flag `experimental_python_types`, the client will convert the results of the query to the 
+corresponding Python types. For example, if the query returns a `DECIMAL` column, the result will be a `Decimal` object.
+
+Limitations of the Python types are described in the 
+[Python types documentation](https://docs.python.org/3/library/datatypes.html). These limitations will generate an 
+exception `trino.exceptions.DataError` if the query returns a value that cannot be converted to the corresponding Python 
+type.
+
+```python
+import trino
+import pytz
+from datetime import datetime
+
+conn = trino.dbapi.connect(
+    ...
+)
+
+cur = conn.cursor(experimental_python_types=True)
+
+params = datetime(2020, 1, 1, 16, 43, 22, 320000, tzinfo=pytz.timezone('America/Los_Angeles'))
+
+cur.execute("SELECT ?", params=(params,))
+rows = cur.fetchall()
+
+assert rows[0][0] == params
+assert cur.description[0][1] == "timestamp with time zone"
+```
+
+# Development
+
+## Getting Started With Development
 
 Start by forking the repository and then modify the code in your fork.
 
diff --git a/setup.py b/setup.py
index 62c74549..01753d0a 100755
--- a/setup.py
+++ b/setup.py
@@ -28,7 +28,7 @@
 kerberos_require = ["requests_kerberos"]
 sqlalchemy_require = ["sqlalchemy~=1.3"]
 
-all_require = kerberos_require + sqlalchemy_require
+all_require = ["pytz"] + kerberos_require + sqlalchemy_require
 
 tests_require = all_require + [
     # httpretty >= 1.1 duplicates requests in `httpretty.latest_requests`
@@ -36,7 +36,6 @@
     "httpretty < 1.1",
     "pytest",
     "pytest-runner",
-    "pytz",
     "click",
 ]
 
diff --git a/tests/integration/test_dbapi_integration.py b/tests/integration/test_dbapi_integration.py
index 4d3d08e4..36ccaf06 100644
--- a/tests/integration/test_dbapi_integration.py
+++ b/tests/integration/test_dbapi_integration.py
@@ -10,7 +10,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import math
-from datetime import datetime
+from datetime import datetime, time, date, timezone, timedelta
+from decimal import Decimal
 
 import pytest
 import pytz
@@ -123,22 +124,335 @@ def test_string_query_param(trino_connection):
     assert rows[0][0] == "six'"
 
 
-def test_datetime_query_param(trino_connection):
+def test_python_types_not_used_when_experimental_python_types_is_not_set(trino_connection):
     cur = trino_connection.cursor()
 
-    cur.execute("SELECT ?", params=(datetime(2020, 1, 1, 0, 0, 0),))
+    cur.execute("""
+    SELECT
+        DECIMAL '0.142857',
+        DATE '2018-01-01',
+        TIMESTAMP '2019-01-01 00:00:00.000+01:00',
+        TIMESTAMP '2019-01-01 00:00:00.000 UTC',
+        TIMESTAMP '2019-01-01 00:00:00.000',
+        TIME '00:00:00.000'
+    """)
+    rows = cur.fetchall()
+
+    for value in rows[0]:
+        assert isinstance(value, str)
+
+    assert rows[0][0] == '0.142857'
+    assert rows[0][1] == '2018-01-01'
+    assert rows[0][2] == '2019-01-01 00:00:00.000 +01:00'
+    assert rows[0][3] == '2019-01-01 00:00:00.000 UTC'
+    assert rows[0][4] == '2019-01-01 00:00:00.000'
+    assert rows[0][5] == '00:00:00.000'
+
+
+def test_decimal_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT ?", params=(Decimal('0.142857'),))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == Decimal('0.142857')
+
+
+def test_null_decimal(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT CAST(NULL AS DECIMAL)")
     rows = cur.fetchall()
 
-    assert rows[0][0] == "2020-01-01 00:00:00.000"
+    assert rows[0][0] is None
+
+
+def test_biggest_decimal(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = Decimal('99999999999999999999999999999999999999')
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
 
-    cur.execute("SELECT ?",
-                params=(datetime(2020, 1, 1, 0, 0, 0, tzinfo=pytz.utc),))
+
+def test_smallest_decimal(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = Decimal('-99999999999999999999999999999999999999')
+    cur.execute("SELECT ?", params=(params,))
     rows = cur.fetchall()
 
-    assert rows[0][0] == "2020-01-01 00:00:00.000 UTC"
+    assert rows[0][0] == params
+
+
+def test_highest_precision_decimal(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = Decimal('0.99999999999999999999999999999999999999')
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
+
+
+def test_datetime_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = datetime(2020, 1, 1, 16, 43, 22, 320000)
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
+    assert cur.description[0][1] == "timestamp"
+
+
+def test_datetime_with_utc_time_zone_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = datetime(2020, 1, 1, 16, 43, 22, 320000, tzinfo=pytz.timezone('UTC'))
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
     assert cur.description[0][1] == "timestamp with time zone"
 
 
+def test_datetime_with_numeric_offset_time_zone_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    tz = timezone(-timedelta(hours=5, minutes=30))
+
+    params = datetime(2020, 1, 1, 16, 43, 22, 320000, tzinfo=tz)
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
+    assert cur.description[0][1] == "timestamp with time zone"
+
+
+def test_datetime_with_named_time_zone_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = datetime(2020, 1, 1, 16, 43, 22, 320000, tzinfo=pytz.timezone('America/Los_Angeles'))
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
+    assert cur.description[0][1] == "timestamp with time zone"
+
+
+def test_datetime_with_trailing_zeros(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT TIMESTAMP '2001-08-22 03:04:05.321000'")
+    rows = cur.fetchall()
+
+    assert rows[0][0] == datetime.strptime("2001-08-22 03:04:05.321000", "%Y-%m-%d %H:%M:%S.%f")
+
+
+def test_null_datetime_with_time_zone(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT CAST(NULL AS TIMESTAMP WITH TIME ZONE)")
+    rows = cur.fetchall()
+
+    assert rows[0][0] is None
+
+
+def test_datetime_with_time_zone_numeric_offset(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT TIMESTAMP '2001-08-22 03:04:05.321 -08:00'")
+    rows = cur.fetchall()
+
+    assert rows[0][0] == datetime.strptime("2001-08-22 03:04:05.321 -08:00", "%Y-%m-%d %H:%M:%S.%f %z")
+
+
+def test_datetimes_with_time_zone_in_dst_gap_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    # This is a datetime that lies within a DST transition and not actually exists.
+    params = datetime(2021, 3, 28, 2, 30, 0, tzinfo=pytz.timezone('Europe/Brussels'))
+    with pytest.raises(trino.exceptions.TrinoUserError):
+        cur.execute("SELECT ?", params=(params,))
+        cur.fetchall()
+
+
+def test_doubled_datetimes(trino_connection):
+    # Trino doesn't distinguish between doubled datetimes that lie within a DST transition. See also
+    # See also https://github.com/trinodb/trino/issues/5781
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = pytz.timezone('US/Eastern').localize(datetime(2002, 10, 27, 1, 30, 0), is_dst=True)
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == datetime(2002, 10, 27, 1, 30, 0, tzinfo=pytz.timezone('US/Eastern'))
+
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = pytz.timezone('US/Eastern').localize(datetime(2002, 10, 27, 1, 30, 0), is_dst=False)
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == datetime(2002, 10, 27, 1, 30, 0, tzinfo=pytz.timezone('US/Eastern'))
+
+
+def test_date_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = datetime(2020, 1, 1, 0, 0, 0).date()
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
+
+
+def test_null_date(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT CAST(NULL AS DATE)")
+    rows = cur.fetchall()
+
+    assert rows[0][0] is None
+
+
+def test_unsupported_python_dates(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    # dates below python min (1-1-1) or above max date (9999-12-31) are not supported
+    for unsupported_date in [
+        '-0001-01-01',
+        '0000-01-01',
+        '10000-01-01',
+        '-4999999-01-01',  # Trino min date
+        '5000000-12-31',  # Trino max date
+    ]:
+        with pytest.raises(trino.exceptions.TrinoDataError):
+            cur.execute(f"SELECT DATE '{unsupported_date}'")
+            cur.fetchall()
+
+
+def test_supported_special_dates_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    for params in (
+            # min python date
+            date(1, 1, 1),
+            # before julian->gregorian switch
+            date(1500, 1, 1),
+            # During julian->gregorian switch
+            date(1752, 9, 4),
+            # before epoch
+            date(1952, 4, 3),
+            date(1970, 1, 1),
+            date(1970, 2, 3),
+            # summer on northern hemisphere (possible DST)
+            date(2017, 7, 1),
+            # winter on northern hemisphere (possible DST on southern hemisphere)
+            date(2017, 1, 1),
+            # winter on southern hemisphere (possible DST on northern hemisphere)
+            date(2017, 12, 31),
+            date(1983, 4, 1),
+            date(1983, 10, 1),
+            # max python date
+            date(9999, 12, 31),
+    ):
+        cur.execute("SELECT ?", params=(params,))
+        rows = cur.fetchall()
+
+        assert rows[0][0] == params
+
+
+def test_time_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = time(12, 3, 44, 333000)
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
+
+
+def test_time_with_named_time_zone_query_param(trino_connection):
+    with pytest.raises(trino.exceptions.NotSupportedError):
+        cur = trino_connection.cursor()
+
+        params = time(16, 43, 22, 320000, tzinfo=pytz.timezone('Asia/Shanghai'))
+
+        cur.execute("SELECT ?", params=(params,))
+
+
+def test_time_with_numeric_offset_time_zone_query_param(trino_connection):
+    with pytest.raises(trino.exceptions.NotSupportedError):
+        cur = trino_connection.cursor()
+
+        tz = timezone(-timedelta(hours=8, minutes=0))
+
+        params = time(16, 43, 22, 320000, tzinfo=tz)
+
+        cur.execute("SELECT ?", params=(params,))
+
+
+def test_time(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT TIME '01:02:03.456'")
+    rows = cur.fetchall()
+
+    assert rows[0][0] == time(1, 2, 3, 456000)
+
+
+def test_null_time(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT CAST(NULL AS TIME)")
+    rows = cur.fetchall()
+
+    assert rows[0][0] is None
+
+
+def test_time_with_time_zone_negative_offset(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT TIME '01:02:03.456 -08:00'")
+    rows = cur.fetchall()
+
+    tz = timezone(-timedelta(hours=8, minutes=0))
+
+    assert rows[0][0] == time(1, 2, 3, 456000, tzinfo=tz)
+
+
+def test_time_with_time_zone_positive_offset(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT TIME '01:02:03.456 +08:00'")
+    rows = cur.fetchall()
+
+    tz = timezone(timedelta(hours=8, minutes=0))
+
+    assert rows[0][0] == time(1, 2, 3, 456000, tzinfo=tz)
+
+
+def test_null_date_with_time_zone(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    cur.execute("SELECT CAST(NULL AS TIME WITH TIME ZONE)")
+    rows = cur.fetchall()
+
+    assert rows[0][0] is None
+
+
 def test_array_query_param(trino_connection):
     cur = trino_connection.cursor()
 
@@ -158,6 +472,70 @@ def test_array_query_param(trino_connection):
     assert rows[0][0] == "array(integer)"
 
 
+def test_array_none_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = [None, None]
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
+
+    cur.execute("SELECT TYPEOF(?)", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == "array(unknown)"
+
+
+def test_array_none_and_another_type_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = [None, 1]
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
+
+    cur.execute("SELECT TYPEOF(?)", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == "array(integer)"
+
+
+def test_array_timestamp_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = [datetime(2020, 1, 1, 0, 0, 0), datetime(2020, 1, 2, 0, 0, 0)]
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
+
+    cur.execute("SELECT TYPEOF(?)", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == "array(timestamp(6))"
+
+
+def test_array_timestamp_with_timezone_query_param(trino_connection):
+    cur = trino_connection.cursor(experimental_python_types=True)
+
+    params = [datetime(2020, 1, 1, 0, 0, 0, tzinfo=pytz.utc), datetime(2020, 1, 2, 0, 0, 0, tzinfo=pytz.utc)]
+
+    cur.execute("SELECT ?", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == params
+
+    cur.execute("SELECT TYPEOF(?)", params=(params,))
+    rows = cur.fetchall()
+
+    assert rows[0][0] == "array(timestamp(6) with time zone)"
+
+
 def test_dict_query_param(trino_connection):
     cur = trino_connection.cursor()
 
diff --git a/trino/client.py b/trino/client.py
index 3142e86f..eb127d3e 100644
--- a/trino/client.py
+++ b/trino/client.py
@@ -36,9 +36,12 @@
 import copy
 import os
 import re
-from typing import Any, Dict, List, Optional, Tuple, Union
 import urllib.parse
+from datetime import datetime, timedelta, timezone
+from decimal import Decimal
+from typing import Any, Dict, List, Optional, Tuple, Union
 
+import pytz
 import requests
 
 import trino.logging
@@ -472,10 +475,11 @@ class TrinoResult(object):
     https://docs.python.org/3/library/stdtypes.html#generator-types
     """
 
-    def __init__(self, query, rows=None):
+    def __init__(self, query, rows=None, experimental_python_types: bool = False):
         self._query = query
         self._rows = rows or []
         self._rownumber = 0
+        self._experimental_python_types = experimental_python_types
 
     @property
     def rownumber(self) -> int:
@@ -494,20 +498,70 @@ def __iter__(self):
             for row in rows:
                 self._rownumber += 1
                 logger.debug("row %s", row)
-                yield row
+                if not self._experimental_python_types:
+                    yield row
+                else:
+                    yield self._map_to_python_types(row, self._query.columns)
 
     @property
     def response_headers(self):
         return self._query.response_headers
 
+    @classmethod
+    def _map_to_python_type(cls, item: Tuple[Any, Dict]) -> Any:
+        (value, data_type) = item
+
+        if value is None:
+            return None
+
+        raw_type = data_type["typeSignature"]["rawType"]
+
+        try:
+            if isinstance(value, list):
+                raw_type = {
+                    "typeSignature": data_type["typeSignature"]["arguments"][0]["value"]
+                }
+                return [cls._map_to_python_type((array_item, raw_type)) for array_item in value]
+            elif "decimal" in raw_type:
+                return Decimal(value)
+            elif raw_type == "date":
+                return datetime.strptime(value, "%Y-%m-%d").date()
+            elif raw_type == "timestamp with time zone":
+                dt, tz = value.rsplit(' ', 1)
+                if tz.startswith('+') or tz.startswith('-'):
+                    return datetime.strptime(value, "%Y-%m-%d %H:%M:%S.%f %z")
+                return datetime.strptime(dt, "%Y-%m-%d %H:%M:%S.%f").replace(tzinfo=pytz.timezone(tz))
+            elif "timestamp" in raw_type:
+                return datetime.strptime(value, "%Y-%m-%d %H:%M:%S.%f")
+            elif "time with time zone" in raw_type:
+                matches = re.match(r'^(.*)([\+\-])(\d{2}):(\d{2})$', value)
+                assert matches is not None
+                assert len(matches.groups()) == 4
+                if matches.group(2) == '-':
+                    tz = -timedelta(hours=int(matches.group(3)), minutes=int(matches.group(4)))
+                else:
+                    tz = timedelta(hours=int(matches.group(3)), minutes=int(matches.group(4)))
+                return datetime.strptime(matches.group(1), "%H:%M:%S.%f").time().replace(tzinfo=timezone(tz))
+            elif "time" in raw_type:
+                return datetime.strptime(value, "%H:%M:%S.%f").time()
+            else:
+                return value
+        except ValueError as e:
+            error_str = f"Could not convert '{value}' into the associated python type for '{raw_type}'"
+            raise trino.exceptions.TrinoDataError(error_str) from e
+
+    def _map_to_python_types(self, row: List[Any], columns: List[Dict[str, Any]]) -> List[Any]:
+        return list(map(self._map_to_python_type, zip(row, columns)))
+
 
 class TrinoQuery(object):
     """Represent the execution of a SQL statement by Trino."""
 
     def __init__(
-        self,
-        request: TrinoRequest,
-        sql: str,
+            self,
+            request: TrinoRequest,
+            sql: str,
+            experimental_python_types: bool = False,
     ) -> None:
         self.query_id: Optional[str] = None
 
@@ -519,8 +573,9 @@ def __init__(
         self._cancelled = False
         self._request = request
         self._sql = sql
-        self._result = TrinoResult(self)
+        self._result = TrinoResult(self, experimental_python_types=experimental_python_types)
         self._response_headers = None
+        self._experimental_python_types = experimental_python_types
 
     @property
     def columns(self):
@@ -567,7 +622,7 @@ def execute(self, additional_http_headers=None) -> TrinoResult:
         self._warnings = getattr(status, "warnings", [])
         if status.next_uri is None:
             self._finished = True
-        self._result = TrinoResult(self, status.rows)
+        self._result = TrinoResult(self, status.rows, self._experimental_python_types)
         return self._result
 
     def _update_state(self, status):
diff --git a/trino/dbapi.py b/trino/dbapi.py
index 1f619335..0b81877c 100644
--- a/trino/dbapi.py
+++ b/trino/dbapi.py
@@ -17,7 +17,7 @@
 Fetch methods returns rows as a list of lists on purpose to let the caller
 decide to convert then to a list of tuples.
 """
-
+from decimal import Decimal
 from typing import Any, List, Optional  # NOQA for mypy types
 
 import copy
@@ -196,7 +196,7 @@ def _create_request(self):
             self.request_timeout,
         )
 
-    def cursor(self):
+    def cursor(self, experimental_python_types=False):
         """Return a new :py:class:`Cursor` object using the connection."""
         if self.isolation_level != IsolationLevel.AUTOCOMMIT:
             if self.transaction is None:
@@ -206,7 +206,7 @@ def cursor(self):
             request = self.transaction._request
         else:
             request = self._create_request()
-        return Cursor(self, request)
+        return Cursor(self, request, experimental_python_types)
 
 
 class Cursor(object):
@@ -217,7 +217,7 @@ class Cursor(object):
 
     """
 
-    def __init__(self, connection, request):
+    def __init__(self, connection, request, experimental_python_types: bool = False):
         if not isinstance(connection, Connection):
             raise ValueError(
                 "connection must be a Connection object: {}".format(type(connection))
@@ -228,6 +228,7 @@ def __init__(self, connection, request):
         self.arraysize = 1
         self._iterator = None
         self._query = None
+        self._experimental_pyton_types = experimental_python_types
 
     def __iter__(self):
         return self._iterator
@@ -305,7 +306,8 @@ def _prepare_statement(self, operation, statement_name):
 
         # Send prepare statement. Copy the _request object to avoid poluting the
         # one that is going to be used to execute the actual operation.
-        query = trino.client.TrinoQuery(copy.deepcopy(self._request), sql=sql)
+        query = trino.client.TrinoQuery(copy.deepcopy(self._request), sql=sql,
+                                        experimental_python_types=self._experimental_pyton_types)
         result = query.execute()
 
         # Iterate until the 'X-Trino-Added-Prepare' header is found or
@@ -327,7 +329,7 @@ def _get_added_prepare_statement_trino_query(
 
         # No need to deepcopy _request here because this is the actual request
         # operation
-        return trino.client.TrinoQuery(self._request, sql=sql)
+        return trino.client.TrinoQuery(self._request, sql=sql, experimental_python_types=self._experimental_pyton_types)
 
     def _format_prepared_param(self, param):
         """
@@ -359,12 +361,27 @@ def _format_prepared_param(self, param):
         if isinstance(param, bytes):
             return "X'%s'" % param.hex()
 
-        if isinstance(param, datetime.datetime):
-            datetime_str = param.strftime("%Y-%m-%d %H:%M:%S.%f %Z")
-            # strip trailing whitespace if param has no zone
-            datetime_str = datetime_str.rstrip(" ")
+        if isinstance(param, datetime.datetime) and param.tzinfo is None:
+            datetime_str = param.strftime("%Y-%m-%d %H:%M:%S.%f")
             return "TIMESTAMP '%s'" % datetime_str
 
+        if isinstance(param, datetime.datetime) and param.tzinfo is not None:
+            datetime_str = param.strftime("%Y-%m-%d %H:%M:%S.%f")
+            # named timezones
+            if hasattr(param.tzinfo, 'zone'):
+                return "TIMESTAMP '%s %s'" % (datetime_str, param.tzinfo.zone)
+            # offset-based timezones
+            return "TIMESTAMP '%s %s'" % (datetime_str, param.tzinfo.tzname(param))
+
+        # We can't calculate the offset for a time without a point in time
+        if isinstance(param, datetime.time) and param.tzinfo is None:
+            time_str = param.strftime("%H:%M:%S.%f")
+            return "TIME '%s'" % time_str
+
+        if isinstance(param, datetime.date):
+            date_str = param.strftime("%Y-%m-%d")
+            return "DATE '%s'" % date_str
+
         if isinstance(param, list):
             return "ARRAY[%s]" % ','.join(map(self._format_prepared_param, param))
 
@@ -379,6 +396,9 @@ def _format_prepared_param(self, param):
         if isinstance(param, uuid.UUID):
             return "UUID '%s'" % param
 
+        if isinstance(param, Decimal):
+            return "DECIMAL '%s'" % param
+
         raise trino.exceptions.NotSupportedError("Query parameter of type '%s' is not supported." % type(param))
 
     def _deallocate_prepare_statement(self, added_prepare_header, statement_name):
@@ -386,7 +406,8 @@ def _deallocate_prepare_statement(self, added_prepare_header, statement_name):
 
         # Send deallocate statement. Copy the _request object to avoid poluting the
         # one that is going to be used to execute the actual operation.
-        query = trino.client.TrinoQuery(copy.deepcopy(self._request), sql=sql)
+        query = trino.client.TrinoQuery(copy.deepcopy(self._request), sql=sql,
+                                        experimental_python_types=self._experimental_pyton_types)
         result = query.execute(
             additional_http_headers={
                 constants.HEADER_PREPARED_STATEMENT: added_prepare_header
@@ -437,7 +458,8 @@ def execute(self, operation, params=None):
                 self._deallocate_prepare_statement(added_prepare_header, statement_name)
 
         else:
-            self._query = trino.client.TrinoQuery(self._request, sql=operation)
+            self._query = trino.client.TrinoQuery(self._request, sql=operation,
+                                                  experimental_python_types=self._experimental_pyton_types)
             result = self._query.execute()
         self._iterator = iter(result)
         return result
@@ -456,6 +478,7 @@ def fetchone(self) -> Optional[List[Any]]:
         """
 
         try:
+            assert self._iterator is not None
             return next(self._iterator)
         except StopIteration:
             return None
diff --git a/trino/exceptions.py b/trino/exceptions.py
index 21cf855b..80d24813 100644
--- a/trino/exceptions.py
+++ b/trino/exceptions.py
@@ -49,6 +49,10 @@ class TrinoAuthError(Exception):
     pass
 
 
+class TrinoDataError(Exception):
+    pass
+
+
 class TrinoQueryError(Exception):
     def __init__(self, error, query_id=None):
         self._error = error