Fix SQLAlchemy support for Impala on Python 3.10 (cloudera#538)

* Add text() wrapper for metadata queries.

Remove tablename from retrieve columnname results.

* Update sqlalchemy.py

remove tablename from get_columns result.

* replace 'r' in re.sub argument

impala/sqlalchemy.py

@@ -26,6 +26,7 @@
 from sqlalchemy.types import (BOOLEAN, SMALLINT, BIGINT, TIMESTAMP, FLOAT,
                               DECIMAL, Integer, Float, String, CHAR, VARCHAR,
                               DATE)
+from sqlalchemy import text
 
 
 registry.register('impala', 'impala.sqlalchemy', 'ImpalaDialect')
@@ -223,7 +224,7 @@ def initialize(self, connection):
         self.default_schema_name = connection.connection.default_db
 
     def _get_server_version_info(self, connection):
-        raw = connection.execute('select version()').scalar()
+        raw = connection.execute(text('select version()')).scalar()
         v = raw.split()[2]
         m = re.match(r'.*?(\d{1,3})\.(\d{1,3})\.(\d{1,3}).*', v)
         return tuple([int(x) for x in m.group(1, 2, 3) if x is not None])
@@ -242,17 +243,17 @@ def get_table_names(self, connection, schema=None, **kw):
             query += ' IN %s' % escaped_schema
         tables = [
             tup[1] if len(tup) > 1 else tup[0]
-            for tup in connection.execute(query).fetchall()
+            for tup in connection.execute(text(query)).fetchall()
         ]
         return tables
-    
+
     def get_view_names(self, connection, schema=None, **kw):
         # Impala doesn't distinguish between tables and view when calling
         # SHOW TABLES. So return a blank list.
         return []
 
     def get_schema_names(self, connection, **kw):
-        rp = connection.execute("SHOW SCHEMAS")
+        rp = connection.execute(text("SHOW SCHEMAS"))
         return [r[0] for r in rp]
 
     def get_columns(self, connection, table_name, schema=None, **kwargs):
@@ -261,15 +262,16 @@ def get_columns(self, connection, table_name, schema=None, **kwargs):
         if schema is not None:
             name = '%s.%s' % (schema, name)
         query = 'SELECT * FROM %s LIMIT 0' % name
-        cursor = connection.execute(query)
+        cursor = connection.execute(text(query))
         schema = cursor.cursor.description
         # We need to fetch the empty results otherwise these queries remain in
         # flight
         cursor.fetchall()
         column_info = []
+        # `select * from table` results in 'tablename.columnname', so strip off 'tablename.'
         for col in schema:
             column_info.append({
-                'name': col[0],
+                'name': col[0].split('.')[-1].strip() if '.' in col[0] else col[0],
                 'type': _impala_type_to_sqlalchemy_type[col[1]],
                 'nullable': True,
                 'autoincrement': False})