diff --git a/docs/src/pages/docs/Connecting to Databases/elasticsearch.mdx b/docs/src/pages/docs/Connecting to Databases/elasticsearch.mdx
index 1704412baa5d3..720964a367038 100644
--- a/docs/src/pages/docs/Connecting to Databases/elasticsearch.mdx	
+++ b/docs/src/pages/docs/Connecting to Databases/elasticsearch.mdx	
@@ -48,3 +48,23 @@ POST /_aliases
 ```
 
 Then register your table with the alias name logstasg_all
+
+**Time zone**
+
+By default, Superset uses UTC time zone for elasticsearch query. If you need to specify a time zone,
+please edit your Database and enter the settings of your specified time zone in the Other > ENGINE PARAMETERS:
+
+
+```
+{
+    "connect_args": {
+        "time_zone": "Asia/Shanghai"
+    }
+}
+```
+
+Another issue to note about the time zone problem is that before elasticsearch7.8, if you want to convert a string into a `DATETIME` object,
+you need to use the `CAST` function,but this function does not support our `time_zone` setting. So it is recommended to upgrade to the version after elasticsearch7.8.
+After elasticsearch7.8, you can use the `DATETIME_PARSE` function to solve this problem.
+The DATETIME_PARSE function is to support our `time_zone` setting, and here you need to fill in your elasticsearch version number in the Other > VERSION setting.
+the superset will use the `DATETIME_PARSE` function for conversion.
diff --git a/superset/connectors/sqla/models.py b/superset/connectors/sqla/models.py
index 34f1ed142833b..d7fd240e23d81 100644
--- a/superset/connectors/sqla/models.py
+++ b/superset/connectors/sqla/models.py
@@ -264,17 +264,23 @@ def is_temporal(self) -> bool:
     def db_engine_spec(self) -> Type[BaseEngineSpec]:
         return self.table.db_engine_spec
 
+    @property
+    def db_extra(self) -> Dict[str, Any]:
+        return self.table.database.get_extra()
+
     @property
     def type_generic(self) -> Optional[utils.GenericDataType]:
         if self.is_dttm:
             return GenericDataType.TEMPORAL
-        column_spec = self.db_engine_spec.get_column_spec(self.type)
+        column_spec = self.db_engine_spec.get_column_spec(
+            self.type, db_extra=self.db_extra
+        )
         return column_spec.generic_type if column_spec else None
 
     def get_sqla_col(self, label: Optional[str] = None) -> Column:
         label = label or self.column_name
         db_engine_spec = self.db_engine_spec
-        column_spec = db_engine_spec.get_column_spec(self.type)
+        column_spec = db_engine_spec.get_column_spec(self.type, db_extra=self.db_extra)
         type_ = column_spec.sqla_type if column_spec else None
         if self.expression:
             tp = self.table.get_template_processor()
@@ -332,7 +338,9 @@ def get_timestamp_expression(
 
         pdf = self.python_date_format
         is_epoch = pdf in ("epoch_s", "epoch_ms")
-        column_spec = self.db_engine_spec.get_column_spec(self.type)
+        column_spec = self.db_engine_spec.get_column_spec(
+            self.type, db_extra=self.db_extra
+        )
         type_ = column_spec.sqla_type if column_spec else DateTime
         if not self.expression and not time_grain and not is_epoch:
             sqla_col = column(self.column_name, type_=type_)
@@ -357,7 +365,11 @@ def dttm_sql_literal(
         ],
     ) -> str:
         """Convert datetime object to a SQL expression string"""
-        sql = self.db_engine_spec.convert_dttm(self.type, dttm) if self.type else None
+        sql = (
+            self.db_engine_spec.convert_dttm(self.type, dttm, db_extra=self.db_extra)
+            if self.type
+            else None
+        )
 
         if sql:
             return sql
@@ -370,10 +382,8 @@ def dttm_sql_literal(
             utils.TimeRangeEndpoint.INCLUSIVE,
             utils.TimeRangeEndpoint.EXCLUSIVE,
         ):
-            tf = (
-                self.table.database.get_extra()
-                .get("python_date_format_by_column_name", {})
-                .get(self.column_name)
+            tf = self.db_extra.get("python_date_format_by_column_name", {}).get(
+                self.column_name
             )
 
         if tf:
@@ -1523,10 +1533,11 @@ def _normalize_prequery_result_type(
             value = value.item()
 
         column_ = columns_by_name[dimension]
+        db_extra: Dict[str, Any] = self.database.get_extra()
 
         if column_.type and column_.is_temporal and isinstance(value, str):
             sql = self.db_engine_spec.convert_dttm(
-                column_.type, dateutil.parser.parse(value),
+                column_.type, dateutil.parser.parse(value), db_extra=db_extra
             )
 
             if sql:
diff --git a/superset/connectors/sqla/utils.py b/superset/connectors/sqla/utils.py
index 0a95364e777c2..e5209e08dcf68 100644
--- a/superset/connectors/sqla/utils.py
+++ b/superset/connectors/sqla/utils.py
@@ -57,7 +57,9 @@ def get_physical_table_metadata(
                 db_type = db_engine_spec.column_datatype_to_string(
                     col["type"], db_dialect
                 )
-                type_spec = db_engine_spec.get_column_spec(db_type)
+                type_spec = db_engine_spec.get_column_spec(
+                    db_type, db_extra=database.get_extra()
+                )
                 col.update(
                     {
                         "type": db_type,
diff --git a/superset/db_engine_specs/athena.py b/superset/db_engine_specs/athena.py
index 666049bac61f7..a33d08f2bf45c 100644
--- a/superset/db_engine_specs/athena.py
+++ b/superset/db_engine_specs/athena.py
@@ -61,7 +61,9 @@ class AthenaEngineSpec(BaseEngineSpec):
     }
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"from_iso8601_date('{dttm.date().isoformat()}')"
diff --git a/superset/db_engine_specs/base.py b/superset/db_engine_specs/base.py
index 8444fc09e1bdb..93974900e3604 100644
--- a/superset/db_engine_specs/base.py
+++ b/superset/db_engine_specs/base.py
@@ -64,7 +64,6 @@
 from superset.utils import core as utils
 from superset.utils.core import ColumnSpec, GenericDataType
 from superset.utils.hashing import md5_sha_from_str
-from superset.utils.memoized import memoized
 from superset.utils.network import is_hostname_valid, is_port_open
 
 if TYPE_CHECKING:
@@ -692,13 +691,14 @@ def df_to_sql(
 
     @classmethod
     def convert_dttm(  # pylint: disable=unused-argument
-        cls, target_type: str, dttm: datetime,
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
     ) -> Optional[str]:
         """
         Convert Python datetime object to a SQL expression
 
         :param target_type: The target type of expression
         :param dttm: The datetime object
+        :param db_extra: The database extra object
         :return: The SQL expression
         """
         return None
@@ -1286,10 +1286,10 @@ def is_select_query(cls, parsed_query: ParsedQuery) -> bool:
         return parsed_query.is_select()
 
     @classmethod
-    @memoized
     def get_column_spec(  # pylint: disable=unused-argument
         cls,
         native_type: Optional[str],
+        db_extra: Optional[Dict[str, Any]] = None,
         source: utils.ColumnTypeSource = utils.ColumnTypeSource.GET_TABLE,
         column_type_mappings: Tuple[
             Tuple[
@@ -1304,6 +1304,7 @@ def get_column_spec(  # pylint: disable=unused-argument
         Converts native database type to sqlalchemy column type.
         :param native_type: Native database typee
         :param source: Type coming from the database table or cursor description
+        :param db_extra: The database extra object
         :return: ColumnSpec object
         """
         col_types = cls.get_sqla_column_type(
@@ -1315,7 +1316,7 @@ def get_column_spec(  # pylint: disable=unused-argument
             # using datetimes
             if generic_type == GenericDataType.TEMPORAL:
                 column_type = literal_dttm_type_factory(
-                    column_type, cls, native_type or ""
+                    column_type, cls, native_type or "", db_extra=db_extra or {}
                 )
             is_dttm = generic_type == GenericDataType.TEMPORAL
             return ColumnSpec(
diff --git a/superset/db_engine_specs/bigquery.py b/superset/db_engine_specs/bigquery.py
index c2b92a01ceeea..30e04c4f2fe9b 100644
--- a/superset/db_engine_specs/bigquery.py
+++ b/superset/db_engine_specs/bigquery.py
@@ -186,7 +186,9 @@ class BigQueryEngineSpec(BaseEngineSpec):
     }
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"CAST('{dttm.date().isoformat()}' AS DATE)"
diff --git a/superset/db_engine_specs/clickhouse.py b/superset/db_engine_specs/clickhouse.py
index d9c218820612d..4f34d2a5543c2 100644
--- a/superset/db_engine_specs/clickhouse.py
+++ b/superset/db_engine_specs/clickhouse.py
@@ -16,7 +16,7 @@
 # under the License.
 import logging
 from datetime import datetime
-from typing import Dict, List, Optional, Type, TYPE_CHECKING
+from typing import Any, Dict, List, Optional, Type, TYPE_CHECKING
 
 from urllib3.exceptions import NewConnectionError
 
@@ -72,7 +72,9 @@ def get_dbapi_mapped_exception(cls, exception: Exception) -> Exception:
         return new_exception(str(exception))
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"toDate('{dttm.date().isoformat()}')"
diff --git a/superset/db_engine_specs/crate.py b/superset/db_engine_specs/crate.py
index 67d68c8349ee0..4d934c448c353 100644
--- a/superset/db_engine_specs/crate.py
+++ b/superset/db_engine_specs/crate.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import Optional, TYPE_CHECKING
+from typing import Any, Dict, Optional, TYPE_CHECKING
 
 from superset.db_engine_specs.base import BaseEngineSpec
 from superset.utils import core as utils
@@ -50,7 +50,9 @@ def epoch_ms_to_dttm(cls) -> str:
         return "{col}"
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.TIMESTAMP:
             return f"{dttm.timestamp() * 1000}"
diff --git a/superset/db_engine_specs/databricks.py b/superset/db_engine_specs/databricks.py
index 7f0e44f785564..50ea59bae2895 100644
--- a/superset/db_engine_specs/databricks.py
+++ b/superset/db_engine_specs/databricks.py
@@ -16,7 +16,7 @@
 # under the License.o
 
 from datetime import datetime
-from typing import Optional
+from typing import Any, Dict, Optional
 
 from superset.db_engine_specs.base import BaseEngineSpec
 from superset.db_engine_specs.hive import HiveEngineSpec
@@ -40,8 +40,10 @@ class DatabricksODBCEngineSpec(BaseEngineSpec):
     _time_grain_expressions = HiveEngineSpec._time_grain_expressions
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
-        return HiveEngineSpec.convert_dttm(target_type, dttm)
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
+        return HiveEngineSpec.convert_dttm(target_type, dttm, db_extra=db_extra)
 
     @classmethod
     def epoch_to_dttm(cls) -> str:
diff --git a/superset/db_engine_specs/dremio.py b/superset/db_engine_specs/dremio.py
index 0d69c8a0f6729..fddba00b5fea3 100644
--- a/superset/db_engine_specs/dremio.py
+++ b/superset/db_engine_specs/dremio.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import Optional
+from typing import Any, Dict, Optional
 
 from superset.db_engine_specs.base import BaseEngineSpec
 from superset.utils import core as utils
@@ -43,7 +43,9 @@ def epoch_to_dttm(cls) -> str:
         return "TO_DATE({col})"
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"TO_DATE('{dttm.date().isoformat()}', 'YYYY-MM-DD')"
diff --git a/superset/db_engine_specs/drill.py b/superset/db_engine_specs/drill.py
index 5b61282ac49ff..3c06017fa2daa 100644
--- a/superset/db_engine_specs/drill.py
+++ b/superset/db_engine_specs/drill.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import Optional
+from typing import Any, Dict, Optional
 from urllib import parse
 
 from sqlalchemy.engine.url import URL
@@ -55,7 +55,9 @@ def epoch_ms_to_dttm(cls) -> str:
         return "TO_DATE({col})"
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"TO_DATE('{dttm.date().isoformat()}', 'yyyy-MM-dd')"
diff --git a/superset/db_engine_specs/druid.py b/superset/db_engine_specs/druid.py
index 8f9cd005dd803..d193daf5844bb 100644
--- a/superset/db_engine_specs/druid.py
+++ b/superset/db_engine_specs/druid.py
@@ -96,7 +96,9 @@ def get_extra_params(database: "Database") -> Dict[str, Any]:
         return extra
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"CAST(TIME_PARSE('{dttm.date().isoformat()}') AS DATE)"
diff --git a/superset/db_engine_specs/elasticsearch.py b/superset/db_engine_specs/elasticsearch.py
index 65042ed05c481..12a5e21e225d6 100644
--- a/superset/db_engine_specs/elasticsearch.py
+++ b/superset/db_engine_specs/elasticsearch.py
@@ -14,8 +14,10 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+import logging
 from datetime import datetime
-from typing import Dict, Optional, Type
+from distutils.version import StrictVersion
+from typing import Any, Dict, Optional, Type
 
 from superset.db_engine_specs.base import BaseEngineSpec
 from superset.db_engine_specs.exceptions import (
@@ -25,6 +27,8 @@
 )
 from superset.utils import core as utils
 
+logger = logging.getLogger()
+
 
 class ElasticSearchEngineSpec(BaseEngineSpec):  # pylint: disable=abstract-method
     engine = "elasticsearch"
@@ -59,9 +63,34 @@ def get_dbapi_exception_mapping(cls) -> Dict[Type[Exception], Type[Exception]]:
         }
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
+
+        db_extra = db_extra or {}
         if target_type.upper() == utils.TemporalType.DATETIME:
+            es_version = db_extra.get("version")
+            # The elasticsearch CAST function does not take effect for the time zone
+            # setting. In elasticsearch7.8 and above, we can use the DATETIME_PARSE
+            # function to solve this problem.
+            supports_dttm_parse = False
+            try:
+                if es_version:
+                    supports_dttm_parse = StrictVersion(es_version) >= StrictVersion(
+                        "7.8"
+                    )
+            except Exception as ex:  # pylint: disable=broad-except
+                logger.error("Unexpected error while convert es_version", exc_info=True)
+                logger.exception(ex)
+
+            if supports_dttm_parse:
+                datetime_formatted = dttm.isoformat(sep=" ", timespec="seconds")
+                return (
+                    f"""DATETIME_PARSE('{datetime_formatted}', 'yyyy-MM-dd HH:mm:ss')"""
+                )
+
             return f"""CAST('{dttm.isoformat(timespec="seconds")}' AS DATETIME)"""
+
         return None
 
 
@@ -87,7 +116,9 @@ class OpenDistroEngineSpec(BaseEngineSpec):  # pylint: disable=abstract-method
     engine_name = "ElasticSearch (OpenDistro SQL)"
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         if target_type.upper() == utils.TemporalType.DATETIME:
             return f"""'{dttm.isoformat(timespec="seconds")}'"""
         return None
diff --git a/superset/db_engine_specs/firebird.py b/superset/db_engine_specs/firebird.py
index 72b462ab45605..9254a3f2aa314 100644
--- a/superset/db_engine_specs/firebird.py
+++ b/superset/db_engine_specs/firebird.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import Optional
+from typing import Any, Dict, Optional
 
 from superset.db_engine_specs.base import BaseEngineSpec, LimitMethod
 from superset.utils import core as utils
@@ -70,7 +70,9 @@ def epoch_to_dttm(cls) -> str:
         return "DATEADD(second, {col}, CAST('00:00:00' AS TIMESTAMP))"
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.TIMESTAMP:
             dttm_formatted = dttm.isoformat(sep=" ")
diff --git a/superset/db_engine_specs/firebolt.py b/superset/db_engine_specs/firebolt.py
index acef2f26b0233..04f48b612a457 100644
--- a/superset/db_engine_specs/firebolt.py
+++ b/superset/db_engine_specs/firebolt.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import Optional
+from typing import Any, Dict, Optional
 
 from superset.db_engine_specs.base import BaseEngineSpec
 from superset.utils import core as utils
@@ -41,7 +41,9 @@ class FireboltEngineSpec(BaseEngineSpec):
     }
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"CAST('{dttm.date().isoformat()}' AS DATE)"
diff --git a/superset/db_engine_specs/hana.py b/superset/db_engine_specs/hana.py
index b2247cf325394..0cc55d08d3f6a 100644
--- a/superset/db_engine_specs/hana.py
+++ b/superset/db_engine_specs/hana.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import Optional
+from typing import Any, Dict, Optional
 
 from superset.db_engine_specs.base import LimitMethod
 from superset.db_engine_specs.postgres import PostgresBaseEngineSpec
@@ -43,7 +43,9 @@ class HanaEngineSpec(PostgresBaseEngineSpec):
     }
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"TO_DATE('{dttm.date().isoformat()}', 'YYYY-MM-DD')"
diff --git a/superset/db_engine_specs/hive.py b/superset/db_engine_specs/hive.py
index f2487c8a638c7..28ace6d93922c 100644
--- a/superset/db_engine_specs/hive.py
+++ b/superset/db_engine_specs/hive.py
@@ -248,7 +248,9 @@ def _get_hive_type(dtype: np.dtype) -> str:
             )
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"CAST('{dttm.date().isoformat()}' AS DATE)"
diff --git a/superset/db_engine_specs/impala.py b/superset/db_engine_specs/impala.py
index f391289ac1d9f..048588c046fd4 100644
--- a/superset/db_engine_specs/impala.py
+++ b/superset/db_engine_specs/impala.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import List, Optional
+from typing import Any, Dict, List, Optional
 
 from sqlalchemy.engine.reflection import Inspector
 
@@ -45,7 +45,9 @@ def epoch_to_dttm(cls) -> str:
         return "from_unixtime({col})"
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"CAST('{dttm.date().isoformat()}' AS DATE)"
diff --git a/superset/db_engine_specs/kylin.py b/superset/db_engine_specs/kylin.py
index 2e59e3100c0f2..dc3836c7373e4 100644
--- a/superset/db_engine_specs/kylin.py
+++ b/superset/db_engine_specs/kylin.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import Optional
+from typing import Any, Dict, Optional
 
 from superset.db_engine_specs.base import BaseEngineSpec
 from superset.utils import core as utils
@@ -40,7 +40,9 @@ class KylinEngineSpec(BaseEngineSpec):  # pylint: disable=abstract-method
     }
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"CAST('{dttm.date().isoformat()}' AS DATE)"
diff --git a/superset/db_engine_specs/mssql.py b/superset/db_engine_specs/mssql.py
index 0572319b2cffd..9db02186981ff 100644
--- a/superset/db_engine_specs/mssql.py
+++ b/superset/db_engine_specs/mssql.py
@@ -98,7 +98,9 @@ def epoch_to_dttm(cls) -> str:
         return "dateadd(S, {col}, '1970-01-01')"
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"CONVERT(DATE, '{dttm.date().isoformat()}', 23)"
diff --git a/superset/db_engine_specs/mysql.py b/superset/db_engine_specs/mysql.py
index fbac330696b7b..0d7493c386e2a 100644
--- a/superset/db_engine_specs/mysql.py
+++ b/superset/db_engine_specs/mysql.py
@@ -151,7 +151,9 @@ class MySQLEngineSpec(BaseEngineSpec, BasicParametersMixin):
     }
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"STR_TO_DATE('{dttm.date().isoformat()}', '%Y-%m-%d')"
@@ -204,6 +206,7 @@ def _extract_error_message(cls, ex: Exception) -> str:
     def get_column_spec(
         cls,
         native_type: Optional[str],
+        db_extra: Optional[Dict[str, Any]] = None,
         source: utils.ColumnTypeSource = utils.ColumnTypeSource.GET_TABLE,
         column_type_mappings: Tuple[
             Tuple[
diff --git a/superset/db_engine_specs/oracle.py b/superset/db_engine_specs/oracle.py
index d91ffb4fc9fcd..ee04e49ffc64d 100644
--- a/superset/db_engine_specs/oracle.py
+++ b/superset/db_engine_specs/oracle.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import Any, List, Optional, Tuple
+from typing import Any, Dict, List, Optional, Tuple
 
 from superset.db_engine_specs.base import BaseEngineSpec, LimitMethod
 from superset.utils import core as utils
@@ -41,7 +41,9 @@ class OracleEngineSpec(BaseEngineSpec):
     }
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"TO_DATE('{dttm.date().isoformat()}', 'YYYY-MM-DD')"
diff --git a/superset/db_engine_specs/postgres.py b/superset/db_engine_specs/postgres.py
index 47cf874ccbc59..3222d00c25266 100644
--- a/superset/db_engine_specs/postgres.py
+++ b/superset/db_engine_specs/postgres.py
@@ -242,7 +242,9 @@ def get_table_names(
         return sorted(tables)
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"TO_DATE('{dttm.date().isoformat()}', 'YYYY-MM-DD')"
@@ -279,6 +281,7 @@ def get_extra_params(database: "Database") -> Dict[str, Any]:
     def get_column_spec(
         cls,
         native_type: Optional[str],
+        db_extra: Optional[Dict[str, Any]] = None,
         source: utils.ColumnTypeSource = utils.ColumnTypeSource.GET_TABLE,
         column_type_mappings: Tuple[
             Tuple[
diff --git a/superset/db_engine_specs/presto.py b/superset/db_engine_specs/presto.py
index 74364300a7fa6..93d39adae2d2f 100644
--- a/superset/db_engine_specs/presto.py
+++ b/superset/db_engine_specs/presto.py
@@ -743,7 +743,9 @@ def adjust_database_uri(
             uri.database = database
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"""from_iso8601_date('{dttm.date().isoformat()}')"""
@@ -1215,6 +1217,7 @@ def is_readonly_query(cls, parsed_query: ParsedQuery) -> bool:
     def get_column_spec(
         cls,
         native_type: Optional[str],
+        db_extra: Optional[Dict[str, Any]] = None,
         source: utils.ColumnTypeSource = utils.ColumnTypeSource.GET_TABLE,
         column_type_mappings: Tuple[
             Tuple[
diff --git a/superset/db_engine_specs/rockset.py b/superset/db_engine_specs/rockset.py
index 5cbf7e686ba23..606b860a5e642 100644
--- a/superset/db_engine_specs/rockset.py
+++ b/superset/db_engine_specs/rockset.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import Optional, TYPE_CHECKING
+from typing import Any, Dict, Optional, TYPE_CHECKING
 
 from superset.db_engine_specs.base import BaseEngineSpec
 from superset.utils import core as utils
@@ -50,7 +50,9 @@ def epoch_ms_to_dttm(cls) -> str:
         return "{col}"
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"DATE '{dttm.date().isoformat()}'"
diff --git a/superset/db_engine_specs/snowflake.py b/superset/db_engine_specs/snowflake.py
index 0310f61d99e1b..058ca89c6af29 100644
--- a/superset/db_engine_specs/snowflake.py
+++ b/superset/db_engine_specs/snowflake.py
@@ -130,7 +130,9 @@ def epoch_ms_to_dttm(cls) -> str:
         return "DATEADD(MS, {col}, '1970-01-01')"
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"TO_DATE('{dttm.date().isoformat()}')"
diff --git a/superset/db_engine_specs/sqlite.py b/superset/db_engine_specs/sqlite.py
index 3b6d8b99ecf99..23512b3cb492f 100644
--- a/superset/db_engine_specs/sqlite.py
+++ b/superset/db_engine_specs/sqlite.py
@@ -97,7 +97,9 @@ def get_all_datasource_names(
         raise Exception(f"Unsupported datasource_type: {datasource_type}")
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt in (utils.TemporalType.TEXT, utils.TemporalType.DATETIME):
             return f"""'{dttm.isoformat(sep=" ", timespec="microseconds")}'"""
diff --git a/superset/db_engine_specs/trino.py b/superset/db_engine_specs/trino.py
index 6491e661ae9a8..8a56e8fb0ad30 100644
--- a/superset/db_engine_specs/trino.py
+++ b/superset/db_engine_specs/trino.py
@@ -46,7 +46,9 @@ class TrinoEngineSpec(BaseEngineSpec):
     }
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             value = dttm.date().isoformat()
diff --git a/superset/models/sql_types/base.py b/superset/models/sql_types/base.py
index 669181b3a4811..597631a372163 100644
--- a/superset/models/sql_types/base.py
+++ b/superset/models/sql_types/base.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 from datetime import datetime
-from typing import Any, Callable, Type, TYPE_CHECKING
+from typing import Any, Callable, Dict, Type, TYPE_CHECKING
 
 from flask_babel import gettext as __
 from sqlalchemy import types
@@ -26,7 +26,10 @@
 
 
 def literal_dttm_type_factory(
-    sqla_type: types.TypeEngine, db_engine_spec: Type["BaseEngineSpec"], col_type: str,
+    sqla_type: types.TypeEngine,
+    db_engine_spec: Type["BaseEngineSpec"],
+    col_type: str,
+    db_extra: Dict[str, Any],
 ) -> types.TypeEngine:
     """
     Create a custom SQLAlchemy type that supports datetime literal binds.
@@ -34,6 +37,7 @@ def literal_dttm_type_factory(
     :param sqla_type: Base type to extend
     :param db_engine_spec: Database engine spec which supports `convert_dttm` method
     :param col_type: native column type as defined in table metadata
+    :param db_extra: The database extra object
     :return: SQLAlchemy type that supports using datetima as literal bind
     """
     # pylint: disable=too-few-public-methods
@@ -42,7 +46,9 @@ class TemporalWrapperType(type(sqla_type)):  # type: ignore
         def literal_processor(self, dialect: Dialect) -> Callable[[Any], Any]:
             def process(value: Any) -> Any:
                 if isinstance(value, datetime):
-                    ts_expression = db_engine_spec.convert_dttm(col_type, value)
+                    ts_expression = db_engine_spec.convert_dttm(
+                        col_type, value, db_extra=db_extra
+                    )
                     if ts_expression is None:
                         raise NotImplementedError(
                             __(
diff --git a/tests/integration_tests/db_engine_specs/base_engine_spec_tests.py b/tests/integration_tests/db_engine_specs/base_engine_spec_tests.py
index 6ee8c4c847359..294c6801e7532 100644
--- a/tests/integration_tests/db_engine_specs/base_engine_spec_tests.py
+++ b/tests/integration_tests/db_engine_specs/base_engine_spec_tests.py
@@ -251,7 +251,7 @@ def test_column_datatype_to_string(self):
 
     def test_convert_dttm(self):
         dttm = self.get_dttm()
-        self.assertIsNone(BaseEngineSpec.convert_dttm("", dttm))
+        self.assertIsNone(BaseEngineSpec.convert_dttm("", dttm, db_extra=None))
 
     def test_pyodbc_rows_to_tuples(self):
         # Test for case when pyodbc.Row is returned (odbc driver)
diff --git a/tests/integration_tests/db_engine_specs/elasticsearch_tests.py b/tests/integration_tests/db_engine_specs/elasticsearch_tests.py
index 92c25aa84351f..7dd5157792acc 100644
--- a/tests/integration_tests/db_engine_specs/elasticsearch_tests.py
+++ b/tests/integration_tests/db_engine_specs/elasticsearch_tests.py
@@ -16,6 +16,7 @@
 # under the License.
 from unittest.mock import MagicMock
 
+import pytest
 from sqlalchemy import column
 
 from superset.db_engine_specs.elasticsearch import (
@@ -26,14 +27,47 @@
 
 
 class TestElasticSearchDbEngineSpec(TestDbEngineSpec):
+    @pytest.fixture(autouse=True)
+    def inject_fixtures(self, caplog):
+        self._caplog = caplog
+
     def test_convert_dttm(self):
         dttm = self.get_dttm()
 
         self.assertEqual(
-            ElasticSearchEngineSpec.convert_dttm("DATETIME", dttm),
+            ElasticSearchEngineSpec.convert_dttm("DATETIME", dttm, db_extra=None),
             "CAST('2019-01-02T03:04:05' AS DATETIME)",
         )
 
+    def test_convert_dttm2(self):
+        """
+        ES 7.8 and above versions need to use the DATETIME_PARSE function to
+        solve the time zone problem
+        """
+        dttm = self.get_dttm()
+        db_extra = {"version": "7.8"}
+
+        self.assertEqual(
+            ElasticSearchEngineSpec.convert_dttm("DATETIME", dttm, db_extra=db_extra),
+            "DATETIME_PARSE('2019-01-02 03:04:05', 'yyyy-MM-dd HH:mm:ss')",
+        )
+
+    def test_convert_dttm3(self):
+        dttm = self.get_dttm()
+        db_extra = {"version": 7.8}
+
+        self.assertEqual(
+            ElasticSearchEngineSpec.convert_dttm("DATETIME", dttm, db_extra=db_extra),
+            "CAST('2019-01-02T03:04:05' AS DATETIME)",
+        )
+
+        self.assertNotEqual(
+            ElasticSearchEngineSpec.convert_dttm("DATETIME", dttm, db_extra=db_extra),
+            "DATETIME_PARSE('2019-01-02 03:04:05', 'yyyy-MM-dd HH:mm:ss')",
+        )
+
+        self.assertIn("Unexpected error while convert es_version", self._caplog.text)
+
     def test_opendistro_convert_dttm(self):
         """
         DB Eng Specs (opendistro): Test convert_dttm
@@ -41,7 +75,7 @@ def test_opendistro_convert_dttm(self):
         dttm = self.get_dttm()
 
         self.assertEqual(
-            OpenDistroEngineSpec.convert_dttm("DATETIME", dttm),
+            OpenDistroEngineSpec.convert_dttm("DATETIME", dttm, db_extra=None),
             "'2019-01-02T03:04:05'",
         )
 
diff --git a/tests/integration_tests/model_tests.py b/tests/integration_tests/model_tests.py
index bc6349c8a2742..c7f7b0ce1b222 100644
--- a/tests/integration_tests/model_tests.py
+++ b/tests/integration_tests/model_tests.py
@@ -565,6 +565,8 @@ def test_data_for_slices_with_query_context(self):
 
 def test_literal_dttm_type_factory():
     orig_type = DateTime()
-    new_type = literal_dttm_type_factory(orig_type, PostgresEngineSpec, "TIMESTAMP")
+    new_type = literal_dttm_type_factory(
+        orig_type, PostgresEngineSpec, "TIMESTAMP", db_extra={}
+    )
     assert type(new_type).__name__ == "TemporalWrapperType"
     assert str(new_type) == str(orig_type)
diff --git a/tests/integration_tests/sqla_models_tests.py b/tests/integration_tests/sqla_models_tests.py
index 03f2d88a0790e..53bf031bf53be 100644
--- a/tests/integration_tests/sqla_models_tests.py
+++ b/tests/integration_tests/sqla_models_tests.py
@@ -521,7 +521,9 @@ def test__normalize_prequery_result_type(
     dimension: str,
     result: Any,
 ) -> None:
-    def _convert_dttm(target_type: str, dttm: datetime) -> Optional[str]:
+    def _convert_dttm(
+        target_type: str, dttm: datetime, db_extra: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
         if target_type.upper() == TemporalType.TIMESTAMP:
             return f"""TIME_PARSE('{dttm.isoformat(timespec="seconds")}')"""