Support result reuse in Athena data sources (#7202)

* Support result reuse

* Update pyathena to 2.25.2

* Separate options

* Regenerate the Poetry lock file

---------

Co-authored-by: SeongTae Jeong <seongtaejg@gmail.com>

Author: dtaniwaki

poetry.lock

@@ -118,7 +118,7 @@ pandas = "1.3.4"
 phoenixdb = "0.7"
 pinotdb = ">=0.4.5"
 protobuf = "3.20.2"
-pyathena = ">=1.5.0,<=1.11.5"
+pyathena = "2.25.2"
 pydgraph = "2.0.2"
 pydruid = "0.5.7"
 pyexasol = "0.12.0"

pyproject.toml

@@ -90,15 +90,26 @@ def configuration_schema(cls):
                     "title": "Athena cost per Tb scanned (USD)",
                     "default": 5,
                 },
+                "result_reuse_enable": {
+                    "type": "boolean",
+                    "title": "Reuse Athena query results",
+                },
+                "result_reuse_minutes": {
+                    "type": "number",
+                    "title": "Minutes to reuse Athena query results",
+                    "default": 60,
+                },
             },
             "required": ["region", "s3_staging_dir"],
-            "extra_options": ["glue", "catalog_ids", "cost_per_tb"],
+            "extra_options": ["glue", "catalog_ids", "cost_per_tb", "result_reuse_enable", "result_reuse_minutes"],
             "order": [
                 "region",
                 "s3_staging_dir",
                 "schema",
                 "work_group",
                 "cost_per_tb",
+                "result_reuse_enable",
+                "result_reuse_minutes",
             ],
             "secret": ["aws_secret_key"],
         }
@@ -247,6 +258,8 @@ def run_query(self, query, user):
             kms_key=self.configuration.get("kms_key", None),
             work_group=self.configuration.get("work_group", "primary"),
             formatter=SimpleFormatter(),
+            result_reuse_enable=self.configuration.get("result_reuse_enable", False),
+            result_reuse_minutes=self.configuration.get("result_reuse_minutes", 60),
             **self._get_iam_credentials(user=user),
         ).cursor()