fix(elasticsearch): time_zone setting does not work for cast datetime expressions

superset/connectors/sqla/models.py

@@ -247,17 +247,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()
@@ -312,7 +318,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_)
@@ -335,7 +343,11 @@ def dttm_sql_literal(
     ) -> str:
         """Convert datetime object to a SQL expression string"""
         dttm_type = self.type or ("DATETIME" if self.is_dttm else None)
-        sql = self.db_engine_spec.convert_dttm(dttm_type, dttm) if dttm_type else None
+        sql = (
+            self.db_engine_spec.convert_dttm(dttm_type, dttm, **self.db_extra)
+            if dttm_type
+            else None
+        )
 
         if sql:
             return sql
@@ -348,10 +360,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:
@@ -1174,7 +1184,9 @@ def get_sqla_query(  # pylint: disable=too-many-arguments,too-many-locals,too-ma
                     sqla_col = col_obj.get_timestamp_expression(filter_grain)
                 else:
                     sqla_col = col_obj.get_sqla_col()
-                col_spec = db_engine_spec.get_column_spec(col_obj.type)
+                col_spec = db_engine_spec.get_column_spec(
+                    col_obj.type, db_extra=self.database.get_extra()
+                )
                 is_list_target = op in (
                     utils.FilterOperator.IN.value,
                     utils.FilterOperator.NOT_IN.value,
@@ -1421,6 +1433,7 @@ def _get_series_orderby(
     def _get_top_groups(
         self, df: pd.DataFrame, dimensions: List[str], groupby_exprs: Dict[str, Any],
     ) -> ColumnElement:
+        db_extra: Dict[str, Any] = self.database.get_extra()
         column_map = {column.column_name: column for column in self.columns}
         groups = []
         for _unused, row in df.iterrows():
@@ -1434,7 +1447,9 @@ def _get_top_groups(
                 # string into a timestamp.
                 if column_map[dimension].is_temporal and isinstance(value, str):
                     dttm = dateutil.parser.parse(value)
-                    value = text(self.db_engine_spec.convert_dttm("TIMESTAMP", dttm))
+                    value = text(
+                        self.db_engine_spec.convert_dttm("TIMESTAMP", dttm, **db_extra)
+                    )
 
                 group.append(groupby_exprs[dimension] == value)
             groups.append(and_(*group))

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,

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, **kwargs: Any
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"from_iso8601_date('{dttm.date().isoformat()}')"

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:
@@ -686,13 +685,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, **kwargs: Any,
     ) -> Optional[str]:
         """
         Convert Python datetime object to a SQL expression
 
         :param target_type: The target type of expression
         :param dttm: The datetime object
+        :param kwargs: The database extra object
         :return: The SQL expression
         """
         return None
@@ -1280,10 +1280,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[
@@ -1298,6 +1298,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(
@@ -1309,7 +1310,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(

superset/db_engine_specs/bigquery.py

@@ -184,7 +184,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, **kwargs: Any
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"CAST('{dttm.date().isoformat()}' AS DATE)"

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, **kwargs: Any
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"toDate('{dttm.date().isoformat()}')"

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, 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, **kwargs: Any
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.TIMESTAMP:
             return f"{dttm.timestamp() * 1000}"

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, Optional
 
 from superset.db_engine_specs.base import BaseEngineSpec
 from superset.db_engine_specs.hive import HiveEngineSpec
@@ -40,7 +40,9 @@ class DatabricksODBCEngineSpec(BaseEngineSpec):
     _time_grain_expressions = HiveEngineSpec._time_grain_expressions
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, **kwargs: Any
+    ) -> Optional[str]:
         return HiveEngineSpec.convert_dttm(target_type, dttm)
 
     @classmethod

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, 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, **kwargs: Any
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"TO_DATE('{dttm.date().isoformat()}', 'YYYY-MM-DD')"

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, 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, **kwargs: Any
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"TO_DATE('{dttm.date().isoformat()}', 'yyyy-MM-dd')"

superset/db_engine_specs/druid.py

@@ -94,7 +94,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, **kwargs: Any
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"CAST(TIME_PARSE('{dttm.date().isoformat()}') AS DATE)"

superset/db_engine_specs/elasticsearch.py

@@ -15,7 +15,8 @@
 # specific language governing permissions and limitations
 # under the License.
 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 (
@@ -59,10 +60,20 @@ def get_dbapi_exception_mapping(cls) -> Dict[Type[Exception], Type[Exception]]:
         }
 
     @classmethod
-    def convert_dttm(cls, target_type: str, dttm: datetime) -> Optional[str]:
-        if target_type.upper() == utils.TemporalType.DATETIME:
-            return f"""CAST('{dttm.isoformat(timespec="seconds")}' AS DATETIME)"""
-        return None
+    def convert_dttm(
+        cls, target_type: str, dttm: datetime, **kwargs: Any
+    ) -> Optional[str]:
+
+        if target_type.upper() != utils.TemporalType.DATETIME:
+            return None
+
+        es_version = kwargs.get("version")
+
+        if es_version and StrictVersion(es_version) >= StrictVersion("7.8"):
+            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)"""
 
 
 class OpenDistroEngineSpec(BaseEngineSpec):  # pylint: disable=abstract-method
@@ -87,7 +98,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, **kwargs: Any
+    ) -> Optional[str]:
         if target_type.upper() == utils.TemporalType.DATETIME:
             return f"""'{dttm.isoformat(timespec="seconds")}'"""
         return None

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, 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, **kwargs: Any
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.TIMESTAMP:
             dttm_formatted = dttm.isoformat(sep=" ")

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, 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, **kwargs: Any
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"CAST('{dttm.date().isoformat()}' AS DATE)"

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, 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, **kwargs: Any
+    ) -> Optional[str]:
         tt = target_type.upper()
         if tt == utils.TemporalType.DATE:
             return f"TO_DATE('{dttm.date().isoformat()}', 'YYYY-MM-DD')"