We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
bug(impala) A delimited table should be explicitly stored as textfile
https://github.com/ibis-project/ibis/blob/88ffe3367cb6a34936e578f6a9b68dc30d559507/ibis/backends/impala/ddl.py#L67
when the cluster's default format is set as parquet, this will cause an exception. It should be explicitly stored as textfile. such as

```python
        if self.lineterminator is not None:
            yield f"LINES TERMINATED BY '{self.lineterminator}'"

        yield 'STORED AS TEXTFILE'
        yield f"LOCATION '{self.path}'"
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ibis/backends/impala/ddl.py`
Content:
```
1 # Copyright 2014 Cloudera Inc.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 # http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import json
16 
17 from ibis.backends.base.sql.ddl import (
18     AlterTable,
19     BaseDDL,
20     CreateTable,
21     CreateTableWithSchema,
22     DropFunction,
23     format_partition,
24     format_schema,
25     format_tblproperties,
26 )
27 from ibis.backends.base.sql.registry import type_to_sql_string
28 
29 
30 class CreateTableParquet(CreateTable):
31     def __init__(
32         self,
33         table_name,
34         path,
35         example_file=None,
36         example_table=None,
37         schema=None,
38         external=True,
39         **kwargs,
40     ):
41         super().__init__(
42             table_name,
43             external=external,
44             format='parquet',
45             path=path,
46             **kwargs,
47         )
48         self.example_file = example_file
49         self.example_table = example_table
50         self.schema = schema
51 
52     @property
53     def _pieces(self):
54         if self.example_file is not None:
55             yield f"LIKE PARQUET '{self.example_file}'"
56         elif self.example_table is not None:
57             yield f"LIKE {self.example_table}"
58         elif self.schema is not None:
59             yield format_schema(self.schema)
60         else:
61             raise NotImplementedError
62 
63         yield self._storage()
64         yield self._location()
65 
66 
67 class DelimitedFormat:
68     def __init__(
69         self,
70         path,
71         delimiter=None,
72         escapechar=None,
73         na_rep=None,
74         lineterminator=None,
75     ):
76         self.path = path
77         self.delimiter = delimiter
78         self.escapechar = escapechar
79         self.lineterminator = lineterminator
80         self.na_rep = na_rep
81 
82     def to_ddl(self):
83         yield 'ROW FORMAT DELIMITED'
84 
85         if self.delimiter is not None:
86             yield f"FIELDS TERMINATED BY '{self.delimiter}'"
87 
88         if self.escapechar is not None:
89             yield f"ESCAPED BY '{self.escapechar}'"
90 
91         if self.lineterminator is not None:
92             yield f"LINES TERMINATED BY '{self.lineterminator}'"
93 
94         yield f"LOCATION '{self.path}'"
95 
96         if self.na_rep is not None:
97             props = {'serialization.null.format': self.na_rep}
98             yield format_tblproperties(props)
99 
100 
101 class AvroFormat:
102     def __init__(self, path, avro_schema):
103         self.path = path
104         self.avro_schema = avro_schema
105 
106     def to_ddl(self):
107         yield 'STORED AS AVRO'
108         yield f"LOCATION '{self.path}'"
109 
110         schema = json.dumps(self.avro_schema, indent=2, sort_keys=True)
111         schema = '\n'.join(x.rstrip() for x in schema.splitlines())
112 
113         props = {'avro.schema.literal': schema}
114         yield format_tblproperties(props)
115 
116 
117 class ParquetFormat:
118     def __init__(self, path):
119         self.path = path
120 
121     def to_ddl(self):
122         yield 'STORED AS PARQUET'
123         yield f"LOCATION '{self.path}'"
124 
125 
126 class CreateTableDelimited(CreateTableWithSchema):
127     def __init__(
128         self,
129         table_name,
130         path,
131         schema,
132         delimiter=None,
133         escapechar=None,
134         lineterminator=None,
135         na_rep=None,
136         external=True,
137         **kwargs,
138     ):
139         table_format = DelimitedFormat(
140             path,
141             delimiter=delimiter,
142             escapechar=escapechar,
143             lineterminator=lineterminator,
144             na_rep=na_rep,
145         )
146         super().__init__(
147             table_name, schema, table_format, external=external, **kwargs
148         )
149 
150 
151 class CreateTableAvro(CreateTable):
152     def __init__(self, table_name, path, avro_schema, external=True, **kwargs):
153         super().__init__(table_name, external=external, **kwargs)
154         self.table_format = AvroFormat(path, avro_schema)
155 
156     @property
157     def _pieces(self):
158         yield '\n'.join(self.table_format.to_ddl())
159 
160 
161 class LoadData(BaseDDL):
162 
163     """
164     Generate DDL for LOAD DATA command. Cannot be cancelled
165     """
166 
167     def __init__(
168         self,
169         table_name,
170         path,
171         database=None,
172         partition=None,
173         partition_schema=None,
174         overwrite=False,
175     ):
176         self.table_name = table_name
177         self.database = database
178         self.path = path
179 
180         self.partition = partition
181         self.partition_schema = partition_schema
182 
183         self.overwrite = overwrite
184 
185     def compile(self):
186         overwrite = 'OVERWRITE ' if self.overwrite else ''
187 
188         if self.partition is not None:
189             partition = '\n' + format_partition(
190                 self.partition, self.partition_schema
191             )
192         else:
193             partition = ''
194 
195         scoped_name = self._get_scoped_name(self.table_name, self.database)
196         return "LOAD DATA INPATH '{}' {}INTO TABLE {}{}".format(
197             self.path, overwrite, scoped_name, partition
198         )
199 
200 
201 class PartitionProperties(AlterTable):
202     def __init__(
203         self,
204         table,
205         partition,
206         partition_schema,
207         location=None,
208         format=None,
209         tbl_properties=None,
210         serde_properties=None,
211     ):
212         super().__init__(
213             table,
214             location=location,
215             format=format,
216             tbl_properties=tbl_properties,
217             serde_properties=serde_properties,
218         )
219         self.partition = partition
220         self.partition_schema = partition_schema
221 
222     def _compile(self, cmd, property_prefix=''):
223         part = format_partition(self.partition, self.partition_schema)
224         if cmd:
225             part = f'{cmd} {part}'
226 
227         props = self._format_properties(property_prefix)
228         action = f'{self.table} {part}{props}'
229         return self._wrap_command(action)
230 
231 
232 class AddPartition(PartitionProperties):
233     def __init__(self, table, partition, partition_schema, location=None):
234         super().__init__(table, partition, partition_schema, location=location)
235 
236     def compile(self):
237         return self._compile('ADD')
238 
239 
240 class AlterPartition(PartitionProperties):
241     def compile(self):
242         return self._compile('', 'SET ')
243 
244 
245 class DropPartition(PartitionProperties):
246     def __init__(self, table, partition, partition_schema):
247         super().__init__(table, partition, partition_schema)
248 
249     def compile(self):
250         return self._compile('DROP')
251 
252 
253 class CacheTable(BaseDDL):
254     def __init__(self, table_name, database=None, pool='default'):
255         self.table_name = table_name
256         self.database = database
257         self.pool = pool
258 
259     def compile(self):
260         scoped_name = self._get_scoped_name(self.table_name, self.database)
261         return "ALTER TABLE {} SET CACHED IN '{}'".format(
262             scoped_name, self.pool
263         )
264 
265 
266 class CreateFunction(BaseDDL):
267 
268     _object_type = 'FUNCTION'
269 
270     def __init__(self, func, name=None, database=None):
271         self.func = func
272         self.name = name or func.name
273         self.database = database
274 
275     def _impala_signature(self):
276         scoped_name = self._get_scoped_name(self.name, self.database)
277         input_sig = _impala_input_signature(self.func.inputs)
278         output_sig = type_to_sql_string(self.func.output)
279 
280         return f'{scoped_name}({input_sig}) returns {output_sig}'
281 
282 
283 class CreateUDF(CreateFunction):
284     def compile(self):
285         create_decl = 'CREATE FUNCTION'
286         impala_sig = self._impala_signature()
287         param_line = "location '{}' symbol='{}'".format(
288             self.func.lib_path, self.func.so_symbol
289         )
290         return ' '.join([create_decl, impala_sig, param_line])
291 
292 
293 class CreateUDA(CreateFunction):
294     def compile(self):
295         create_decl = 'CREATE AGGREGATE FUNCTION'
296         impala_sig = self._impala_signature()
297         tokens = [f"location '{self.func.lib_path}'"]
298 
299         fn_names = (
300             'init_fn',
301             'update_fn',
302             'merge_fn',
303             'serialize_fn',
304             'finalize_fn',
305         )
306 
307         for fn in fn_names:
308             value = getattr(self.func, fn)
309             if value is not None:
310                 tokens.append(f"{fn}='{value}'")
311 
312         return ' '.join([create_decl, impala_sig]) + ' ' + '\n'.join(tokens)
313 
314 
315 class DropFunction(DropFunction):
316     def _impala_signature(self):
317         full_name = self._get_scoped_name(self.name, self.database)
318         input_sig = _impala_input_signature(self.inputs)
319         return f'{full_name}({input_sig})'
320 
321 
322 class ListFunction(BaseDDL):
323     def __init__(self, database, like=None, aggregate=False):
324         self.database = database
325         self.like = like
326         self.aggregate = aggregate
327 
328     def compile(self):
329         statement = 'SHOW '
330         if self.aggregate:
331             statement += 'AGGREGATE '
332         statement += f'FUNCTIONS IN {self.database}'
333         if self.like:
334             statement += f" LIKE '{self.like}'"
335         return statement
336 
337 
338 def _impala_input_signature(inputs):
339     # TODO: varargs '{}...'.format(val)
340     return ', '.join(map(type_to_sql_string, inputs))
341
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/ibis/backends/impala/ddl.py b/ibis/backends/impala/ddl.py
--- a/ibis/backends/impala/ddl.py
+++ b/ibis/backends/impala/ddl.py
@@ -91,6 +91,7 @@
         if self.lineterminator is not None:
             yield f"LINES TERMINATED BY '{self.lineterminator}'"
 
+        yield 'STORED AS TEXTFILE'
         yield f"LOCATION '{self.path}'"
 
         if self.na_rep is not None:

{"golden_diff": "diff --git a/ibis/backends/impala/ddl.py b/ibis/backends/impala/ddl.py\n--- a/ibis/backends/impala/ddl.py\n+++ b/ibis/backends/impala/ddl.py\n@@ -91,6 +91,7 @@\n         if self.lineterminator is not None:\n             yield f\"LINES TERMINATED BY '{self.lineterminator}'\"\n \n+        yield 'STORED AS TEXTFILE'\n         yield f\"LOCATION '{self.path}'\"\n \n         if self.na_rep is not None:\n", "issue": "bug(impala) A delimited table should be explicitly stored as textfile\nhttps://github.com/ibis-project/ibis/blob/88ffe3367cb6a34936e578f6a9b68dc30d559507/ibis/backends/impala/ddl.py#L67\r\nwhen the cluster's default format is set as parquet, this will cause an exception. It should be explicitly stored as textfile. such as\r\n\r\n```python\r\n        if self.lineterminator is not None:\r\n            yield f\"LINES TERMINATED BY '{self.lineterminator}'\"\r\n\r\n        yield 'STORED AS TEXTFILE'\r\n        yield f\"LOCATION '{self.path}'\"\r\n```\n", "before_files": [{"content": "# Copyright 2014 Cloudera Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport json\n\nfrom ibis.backends.base.sql.ddl import (\n    AlterTable,\n    BaseDDL,\n    CreateTable,\n    CreateTableWithSchema,\n    DropFunction,\n    format_partition,\n    format_schema,\n    format_tblproperties,\n)\nfrom ibis.backends.base.sql.registry import type_to_sql_string\n\n\nclass CreateTableParquet(CreateTable):\n    def __init__(\n        self,\n        table_name,\n        path,\n        example_file=None,\n        example_table=None,\n        schema=None,\n        external=True,\n        **kwargs,\n    ):\n        super().__init__(\n            table_name,\n            external=external,\n            format='parquet',\n            path=path,\n            **kwargs,\n        )\n        self.example_file = example_file\n        self.example_table = example_table\n        self.schema = schema\n\n    @property\n    def _pieces(self):\n        if self.example_file is not None:\n            yield f\"LIKE PARQUET '{self.example_file}'\"\n        elif self.example_table is not None:\n            yield f\"LIKE {self.example_table}\"\n        elif self.schema is not None:\n            yield format_schema(self.schema)\n        else:\n            raise NotImplementedError\n\n        yield self._storage()\n        yield self._location()\n\n\nclass DelimitedFormat:\n    def __init__(\n        self,\n        path,\n        delimiter=None,\n        escapechar=None,\n        na_rep=None,\n        lineterminator=None,\n    ):\n        self.path = path\n        self.delimiter = delimiter\n        self.escapechar = escapechar\n        self.lineterminator = lineterminator\n        self.na_rep = na_rep\n\n    def to_ddl(self):\n        yield 'ROW FORMAT DELIMITED'\n\n        if self.delimiter is not None:\n            yield f\"FIELDS TERMINATED BY '{self.delimiter}'\"\n\n        if self.escapechar is not None:\n            yield f\"ESCAPED BY '{self.escapechar}'\"\n\n        if self.lineterminator is not None:\n            yield f\"LINES TERMINATED BY '{self.lineterminator}'\"\n\n        yield f\"LOCATION '{self.path}'\"\n\n        if self.na_rep is not None:\n            props = {'serialization.null.format': self.na_rep}\n            yield format_tblproperties(props)\n\n\nclass AvroFormat:\n    def __init__(self, path, avro_schema):\n        self.path = path\n        self.avro_schema = avro_schema\n\n    def to_ddl(self):\n        yield 'STORED AS AVRO'\n        yield f\"LOCATION '{self.path}'\"\n\n        schema = json.dumps(self.avro_schema, indent=2, sort_keys=True)\n        schema = '\\n'.join(x.rstrip() for x in schema.splitlines())\n\n        props = {'avro.schema.literal': schema}\n        yield format_tblproperties(props)\n\n\nclass ParquetFormat:\n    def __init__(self, path):\n        self.path = path\n\n    def to_ddl(self):\n        yield 'STORED AS PARQUET'\n        yield f\"LOCATION '{self.path}'\"\n\n\nclass CreateTableDelimited(CreateTableWithSchema):\n    def __init__(\n        self,\n        table_name,\n        path,\n        schema,\n        delimiter=None,\n        escapechar=None,\n        lineterminator=None,\n        na_rep=None,\n        external=True,\n        **kwargs,\n    ):\n        table_format = DelimitedFormat(\n            path,\n            delimiter=delimiter,\n            escapechar=escapechar,\n            lineterminator=lineterminator,\n            na_rep=na_rep,\n        )\n        super().__init__(\n            table_name, schema, table_format, external=external, **kwargs\n        )\n\n\nclass CreateTableAvro(CreateTable):\n    def __init__(self, table_name, path, avro_schema, external=True, **kwargs):\n        super().__init__(table_name, external=external, **kwargs)\n        self.table_format = AvroFormat(path, avro_schema)\n\n    @property\n    def _pieces(self):\n        yield '\\n'.join(self.table_format.to_ddl())\n\n\nclass LoadData(BaseDDL):\n\n    \"\"\"\n    Generate DDL for LOAD DATA command. Cannot be cancelled\n    \"\"\"\n\n    def __init__(\n        self,\n        table_name,\n        path,\n        database=None,\n        partition=None,\n        partition_schema=None,\n        overwrite=False,\n    ):\n        self.table_name = table_name\n        self.database = database\n        self.path = path\n\n        self.partition = partition\n        self.partition_schema = partition_schema\n\n        self.overwrite = overwrite\n\n    def compile(self):\n        overwrite = 'OVERWRITE ' if self.overwrite else ''\n\n        if self.partition is not None:\n            partition = '\\n' + format_partition(\n                self.partition, self.partition_schema\n            )\n        else:\n            partition = ''\n\n        scoped_name = self._get_scoped_name(self.table_name, self.database)\n        return \"LOAD DATA INPATH '{}' {}INTO TABLE {}{}\".format(\n            self.path, overwrite, scoped_name, partition\n        )\n\n\nclass PartitionProperties(AlterTable):\n    def __init__(\n        self,\n        table,\n        partition,\n        partition_schema,\n        location=None,\n        format=None,\n        tbl_properties=None,\n        serde_properties=None,\n    ):\n        super().__init__(\n            table,\n            location=location,\n            format=format,\n            tbl_properties=tbl_properties,\n            serde_properties=serde_properties,\n        )\n        self.partition = partition\n        self.partition_schema = partition_schema\n\n    def _compile(self, cmd, property_prefix=''):\n        part = format_partition(self.partition, self.partition_schema)\n        if cmd:\n            part = f'{cmd} {part}'\n\n        props = self._format_properties(property_prefix)\n        action = f'{self.table} {part}{props}'\n        return self._wrap_command(action)\n\n\nclass AddPartition(PartitionProperties):\n    def __init__(self, table, partition, partition_schema, location=None):\n        super().__init__(table, partition, partition_schema, location=location)\n\n    def compile(self):\n        return self._compile('ADD')\n\n\nclass AlterPartition(PartitionProperties):\n    def compile(self):\n        return self._compile('', 'SET ')\n\n\nclass DropPartition(PartitionProperties):\n    def __init__(self, table, partition, partition_schema):\n        super().__init__(table, partition, partition_schema)\n\n    def compile(self):\n        return self._compile('DROP')\n\n\nclass CacheTable(BaseDDL):\n    def __init__(self, table_name, database=None, pool='default'):\n        self.table_name = table_name\n        self.database = database\n        self.pool = pool\n\n    def compile(self):\n        scoped_name = self._get_scoped_name(self.table_name, self.database)\n        return \"ALTER TABLE {} SET CACHED IN '{}'\".format(\n            scoped_name, self.pool\n        )\n\n\nclass CreateFunction(BaseDDL):\n\n    _object_type = 'FUNCTION'\n\n    def __init__(self, func, name=None, database=None):\n        self.func = func\n        self.name = name or func.name\n        self.database = database\n\n    def _impala_signature(self):\n        scoped_name = self._get_scoped_name(self.name, self.database)\n        input_sig = _impala_input_signature(self.func.inputs)\n        output_sig = type_to_sql_string(self.func.output)\n\n        return f'{scoped_name}({input_sig}) returns {output_sig}'\n\n\nclass CreateUDF(CreateFunction):\n    def compile(self):\n        create_decl = 'CREATE FUNCTION'\n        impala_sig = self._impala_signature()\n        param_line = \"location '{}' symbol='{}'\".format(\n            self.func.lib_path, self.func.so_symbol\n        )\n        return ' '.join([create_decl, impala_sig, param_line])\n\n\nclass CreateUDA(CreateFunction):\n    def compile(self):\n        create_decl = 'CREATE AGGREGATE FUNCTION'\n        impala_sig = self._impala_signature()\n        tokens = [f\"location '{self.func.lib_path}'\"]\n\n        fn_names = (\n            'init_fn',\n            'update_fn',\n            'merge_fn',\n            'serialize_fn',\n            'finalize_fn',\n        )\n\n        for fn in fn_names:\n            value = getattr(self.func, fn)\n            if value is not None:\n                tokens.append(f\"{fn}='{value}'\")\n\n        return ' '.join([create_decl, impala_sig]) + ' ' + '\\n'.join(tokens)\n\n\nclass DropFunction(DropFunction):\n    def _impala_signature(self):\n        full_name = self._get_scoped_name(self.name, self.database)\n        input_sig = _impala_input_signature(self.inputs)\n        return f'{full_name}({input_sig})'\n\n\nclass ListFunction(BaseDDL):\n    def __init__(self, database, like=None, aggregate=False):\n        self.database = database\n        self.like = like\n        self.aggregate = aggregate\n\n    def compile(self):\n        statement = 'SHOW '\n        if self.aggregate:\n            statement += 'AGGREGATE '\n        statement += f'FUNCTIONS IN {self.database}'\n        if self.like:\n            statement += f\" LIKE '{self.like}'\"\n        return statement\n\n\ndef _impala_input_signature(inputs):\n    # TODO: varargs '{}...'.format(val)\n    return ', '.join(map(type_to_sql_string, inputs))\n", "path": "ibis/backends/impala/ddl.py"}], "after_files": [{"content": "# Copyright 2014 Cloudera Inc.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport json\n\nfrom ibis.backends.base.sql.ddl import (\n    AlterTable,\n    BaseDDL,\n    CreateTable,\n    CreateTableWithSchema,\n    DropFunction,\n    format_partition,\n    format_schema,\n    format_tblproperties,\n)\nfrom ibis.backends.base.sql.registry import type_to_sql_string\n\n\nclass CreateTableParquet(CreateTable):\n    def __init__(\n        self,\n        table_name,\n        path,\n        example_file=None,\n        example_table=None,\n        schema=None,\n        external=True,\n        **kwargs,\n    ):\n        super().__init__(\n            table_name,\n            external=external,\n            format='parquet',\n            path=path,\n            **kwargs,\n        )\n        self.example_file = example_file\n        self.example_table = example_table\n        self.schema = schema\n\n    @property\n    def _pieces(self):\n        if self.example_file is not None:\n            yield f\"LIKE PARQUET '{self.example_file}'\"\n        elif self.example_table is not None:\n            yield f\"LIKE {self.example_table}\"\n        elif self.schema is not None:\n            yield format_schema(self.schema)\n        else:\n            raise NotImplementedError\n\n        yield self._storage()\n        yield self._location()\n\n\nclass DelimitedFormat:\n    def __init__(\n        self,\n        path,\n        delimiter=None,\n        escapechar=None,\n        na_rep=None,\n        lineterminator=None,\n    ):\n        self.path = path\n        self.delimiter = delimiter\n        self.escapechar = escapechar\n        self.lineterminator = lineterminator\n        self.na_rep = na_rep\n\n    def to_ddl(self):\n        yield 'ROW FORMAT DELIMITED'\n\n        if self.delimiter is not None:\n            yield f\"FIELDS TERMINATED BY '{self.delimiter}'\"\n\n        if self.escapechar is not None:\n            yield f\"ESCAPED BY '{self.escapechar}'\"\n\n        if self.lineterminator is not None:\n            yield f\"LINES TERMINATED BY '{self.lineterminator}'\"\n\n        yield 'STORED AS TEXTFILE'\n        yield f\"LOCATION '{self.path}'\"\n\n        if self.na_rep is not None:\n            props = {'serialization.null.format': self.na_rep}\n            yield format_tblproperties(props)\n\n\nclass AvroFormat:\n    def __init__(self, path, avro_schema):\n        self.path = path\n        self.avro_schema = avro_schema\n\n    def to_ddl(self):\n        yield 'STORED AS AVRO'\n        yield f\"LOCATION '{self.path}'\"\n\n        schema = json.dumps(self.avro_schema, indent=2, sort_keys=True)\n        schema = '\\n'.join(x.rstrip() for x in schema.splitlines())\n\n        props = {'avro.schema.literal': schema}\n        yield format_tblproperties(props)\n\n\nclass ParquetFormat:\n    def __init__(self, path):\n        self.path = path\n\n    def to_ddl(self):\n        yield 'STORED AS PARQUET'\n        yield f\"LOCATION '{self.path}'\"\n\n\nclass CreateTableDelimited(CreateTableWithSchema):\n    def __init__(\n        self,\n        table_name,\n        path,\n        schema,\n        delimiter=None,\n        escapechar=None,\n        lineterminator=None,\n        na_rep=None,\n        external=True,\n        **kwargs,\n    ):\n        table_format = DelimitedFormat(\n            path,\n            delimiter=delimiter,\n            escapechar=escapechar,\n            lineterminator=lineterminator,\n            na_rep=na_rep,\n        )\n        super().__init__(\n            table_name, schema, table_format, external=external, **kwargs\n        )\n\n\nclass CreateTableAvro(CreateTable):\n    def __init__(self, table_name, path, avro_schema, external=True, **kwargs):\n        super().__init__(table_name, external=external, **kwargs)\n        self.table_format = AvroFormat(path, avro_schema)\n\n    @property\n    def _pieces(self):\n        yield '\\n'.join(self.table_format.to_ddl())\n\n\nclass LoadData(BaseDDL):\n\n    \"\"\"\n    Generate DDL for LOAD DATA command. Cannot be cancelled\n    \"\"\"\n\n    def __init__(\n        self,\n        table_name,\n        path,\n        database=None,\n        partition=None,\n        partition_schema=None,\n        overwrite=False,\n    ):\n        self.table_name = table_name\n        self.database = database\n        self.path = path\n\n        self.partition = partition\n        self.partition_schema = partition_schema\n\n        self.overwrite = overwrite\n\n    def compile(self):\n        overwrite = 'OVERWRITE ' if self.overwrite else ''\n\n        if self.partition is not None:\n            partition = '\\n' + format_partition(\n                self.partition, self.partition_schema\n            )\n        else:\n            partition = ''\n\n        scoped_name = self._get_scoped_name(self.table_name, self.database)\n        return \"LOAD DATA INPATH '{}' {}INTO TABLE {}{}\".format(\n            self.path, overwrite, scoped_name, partition\n        )\n\n\nclass PartitionProperties(AlterTable):\n    def __init__(\n        self,\n        table,\n        partition,\n        partition_schema,\n        location=None,\n        format=None,\n        tbl_properties=None,\n        serde_properties=None,\n    ):\n        super().__init__(\n            table,\n            location=location,\n            format=format,\n            tbl_properties=tbl_properties,\n            serde_properties=serde_properties,\n        )\n        self.partition = partition\n        self.partition_schema = partition_schema\n\n    def _compile(self, cmd, property_prefix=''):\n        part = format_partition(self.partition, self.partition_schema)\n        if cmd:\n            part = f'{cmd} {part}'\n\n        props = self._format_properties(property_prefix)\n        action = f'{self.table} {part}{props}'\n        return self._wrap_command(action)\n\n\nclass AddPartition(PartitionProperties):\n    def __init__(self, table, partition, partition_schema, location=None):\n        super().__init__(table, partition, partition_schema, location=location)\n\n    def compile(self):\n        return self._compile('ADD')\n\n\nclass AlterPartition(PartitionProperties):\n    def compile(self):\n        return self._compile('', 'SET ')\n\n\nclass DropPartition(PartitionProperties):\n    def __init__(self, table, partition, partition_schema):\n        super().__init__(table, partition, partition_schema)\n\n    def compile(self):\n        return self._compile('DROP')\n\n\nclass CacheTable(BaseDDL):\n    def __init__(self, table_name, database=None, pool='default'):\n        self.table_name = table_name\n        self.database = database\n        self.pool = pool\n\n    def compile(self):\n        scoped_name = self._get_scoped_name(self.table_name, self.database)\n        return \"ALTER TABLE {} SET CACHED IN '{}'\".format(\n            scoped_name, self.pool\n        )\n\n\nclass CreateFunction(BaseDDL):\n\n    _object_type = 'FUNCTION'\n\n    def __init__(self, func, name=None, database=None):\n        self.func = func\n        self.name = name or func.name\n        self.database = database\n\n    def _impala_signature(self):\n        scoped_name = self._get_scoped_name(self.name, self.database)\n        input_sig = _impala_input_signature(self.func.inputs)\n        output_sig = type_to_sql_string(self.func.output)\n\n        return f'{scoped_name}({input_sig}) returns {output_sig}'\n\n\nclass CreateUDF(CreateFunction):\n    def compile(self):\n        create_decl = 'CREATE FUNCTION'\n        impala_sig = self._impala_signature()\n        param_line = \"location '{}' symbol='{}'\".format(\n            self.func.lib_path, self.func.so_symbol\n        )\n        return ' '.join([create_decl, impala_sig, param_line])\n\n\nclass CreateUDA(CreateFunction):\n    def compile(self):\n        create_decl = 'CREATE AGGREGATE FUNCTION'\n        impala_sig = self._impala_signature()\n        tokens = [f\"location '{self.func.lib_path}'\"]\n\n        fn_names = (\n            'init_fn',\n            'update_fn',\n            'merge_fn',\n            'serialize_fn',\n            'finalize_fn',\n        )\n\n        for fn in fn_names:\n            value = getattr(self.func, fn)\n            if value is not None:\n                tokens.append(f\"{fn}='{value}'\")\n\n        return ' '.join([create_decl, impala_sig]) + ' ' + '\\n'.join(tokens)\n\n\nclass DropFunction(DropFunction):\n    def _impala_signature(self):\n        full_name = self._get_scoped_name(self.name, self.database)\n        input_sig = _impala_input_signature(self.inputs)\n        return f'{full_name}({input_sig})'\n\n\nclass ListFunction(BaseDDL):\n    def __init__(self, database, like=None, aggregate=False):\n        self.database = database\n        self.like = like\n        self.aggregate = aggregate\n\n    def compile(self):\n        statement = 'SHOW '\n        if self.aggregate:\n            statement += 'AGGREGATE '\n        statement += f'FUNCTIONS IN {self.database}'\n        if self.like:\n            statement += f\" LIKE '{self.like}'\"\n        return statement\n\n\ndef _impala_input_signature(inputs):\n    # TODO: varargs '{}...'.format(val)\n    return ', '.join(map(type_to_sql_string, inputs))\n", "path": "ibis/backends/impala/ddl.py"}]}