MDEV-5341 ConnectSE: discovery for ODBC tables does not work if tables

with the same names present in multiple schemas

The "TABNAME" option now supports qualified table names,
to connect to tables residing in a particular schema and catalog.

Qualified table names have the following format:

  [[CatalogName.]SchemaName.]TableName

Qualified table names can be used:

1. In "normal" tables:

CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC
CONNECTION='DSN=ConnectEng;UID=mtr;PWD=mtr'
TABNAME='schema1.t1';

2. In catalog tables (CATFUNC=Tables  and CATFUNC=Columns)

CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC
CONNECTION='DSN=postgresql;UID=user;PWD=password'
TABNAME='schema1.t1';

Note, the % and _ wildcards are supported in
the schema name and the table name parts:

CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC
CONNECTION='DSN=postgresql;UID=user;PWD=password'
TABNAME='%.t1';

CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC
CONNECTION='DSN=postgresql;UID=user;PWD=password'
TABNAME='schema1.%';

storage/connect/mysql-test/connect/t/have_odbc_postgresql.inc

+--disable_query_log
+--error 0,ER_UNKNOWN_ERROR
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CATFUNC=Sources;
+if ($mysql_errno)
+{
+  Skip No ODBC support;
+}
+if (!`SELECT count(*) FROM t1 WHERE Name='ConnectEnginePostgresql'`)
+{
+  DROP TABLE t1;
+  Skip Need SQLite3 ODBC Driver;
+}
+SHOW CREATE TABLE t1;
+DROP TABLE t1;
+--enable_query_log

storage/connect/mysql-test/connect/t/odbc_postgresql.sql

+--
+-- The SQL script to create PostgreSQL data for odbc_postgresql.test
+--
+-- Run this script as a admin user:
+-- sudo -u postgres psql < odbc_postgresql.create.sql
+
+DROP DATABASE IF EXISTS mtr;
+DROP USER IF EXISTS mtr;
+
+CREATE USER mtr WITH PASSWORD 'mtr';
+CREATE DATABASE mtr OWNER=mtr ENCODING='UTF8';
+GRANT ALL ON DATABASE mtr TO mtr;
+\c mtr
+SET role mtr;
+CREATE TABLE t1 (a INT NOT NULL);
+INSERT INTO t1 VALUES (10),(20),(30);
+CREATE VIEW v1 AS SELECT * FROM t1;
+CREATE TABLE t2 (a INT NOT NULL);
+INSERT INTO t2 VALUES (40),(50),(60);
+CREATE SCHEMA schema1 AUTHORIZATION mtr;
+CREATE TABLE schema1.t1 (a CHAR(10) NOT NULL);
+INSERT INTO schema1.t1 VALUES ('aaa'),('bbb'),('ccc'),('яяя');
+CREATE VIEW schema1.v1 AS SELECT * FROM schema1.t1;
+CREATE TABLE schema1.t2 (a CHAR(10) NOT NULL);
+INSERT INTO schema1.t2 VALUES ('xxx'),('yyy'),('zzz'),('ÄÖÜ');

storage/connect/mysql-test/connect/t/odbc_postgresql.test

+--source have_odbc_postgresql.inc
+--source include/not_embedded.inc
+
+#
+# To configure your system to be able to run this test,
+# follow through the following steps:
+#
+# 1. Install and configure PostgreSQL database to stat on the system startup
+#
+# 2. Create user, database, schema and tables to be used by mtr:
+#    sudo -u postgres < odbc_postgresql.sql
+#
+# 3. Install PostgreSQL ODBC Driver.
+# - On CentOS, Fedora:
+#   sudo yum install postgresql-odbc
+# - On Ubuntu, Debian:
+#   sudo apt-get install odbc-postgresql
+#
+# 4. Create a data source with the name "ConnectEnginePostgresql"
+# - On Windows: use odbcadm.exe
+# - On Linux: put these lines into /etc/odbc.ini
+#
+#[ConnectEnginePostgresql]
+#Description=PostgreSQL DSN for ConnectSE
+#Driver=PostgreSQL
+#Database=mtr
+#Servername=localhost
+#Port=5432
+#
+
+SET NAMES utf8;
+
+--echo #
+--echo # Checking CATFUNC=Tables
+--echo #
+--echo
+--echo # All tables in all schemas
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Tables;
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # All tables in all schemas
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Tables TABNAME='%.%.%';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # All tables in all schemas
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Tables TABNAME='%.%';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # All tables in the default schema ("public")
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Tables TABNAME='%';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # All tables "t1" in all schemas
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Tables TABNAME='%.%.t1';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # All tables "t1" in all schemas
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Tables TABNAME='%.t1';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # Table "t1" in the default schema ("public")
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Tables TABNAME='t1';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # Table "t1" in the schema "public"
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Tables TABNAME='%.public.t1';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # Table "t1" in the schema "schema1"
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Tables TABNAME='%.schema1.t1';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # All tables "t1" in all schemas (Catalog name is ignored by PostgreSQL)
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Tables TABNAME='xxx.%.t1';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo #
+--echo # Checking CATFUNC=Columns
+--echo #
+--echo
+
+#
+# For some reasons SQLColumn (unlike SQLTables) include columns of system
+# tables from the schemas like "information_schema", "pg_catalog", "pg_toast".
+# So we add the "Table_Owner IN ('public','schema1')" clause into some queries.
+#
+
+--echo # All columns in the schemas "public" and "schema1"
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Columns;
+SELECT * FROM t1 WHERE Table_Owner IN ('public','schema1') ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # All columns in the schemas "public" and "schema1"
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Columns TABNAME='%.%.%';
+SELECT * FROM t1 WHERE Table_Owner IN ('public','schema1') ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # All tables "t1" in all schemas
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Columns TABNAME='%.%.t1';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # Table "t1" in the schema "public"
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Columns TABNAME='%.public.t1';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # Table "t1" in the schema "schema1"
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Columns TABNAME='%.schema1.t1';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+--echo # All tables "t1" in all schemas (Catalog name is ignored by PostgreSQL)
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr' CATFUNC=Columns TABNAME='xxx.%.t1';
+SELECT * FROM t1 ORDER BY Table_Owner, Table_Name;
+DROP TABLE t1;
+
+
+--echo #
+--echo # Checking tables
+--echo #
+--echo
+
+--echo # Table "t1" in the default schema ("public")
+CREATE TABLE t1 ENGINE=CONNECT TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr';
+SHOW CREATE TABLE t1;
+SELECT * FROM t1;
+
+CREATE TABLE t2 AS SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+SELECT * FROM t2;
+DROP TABLE t2;
+
+CREATE VIEW v1 AS SELECT * FROM t1;
+SELECT * FROM v1;
+DROP VIEW v1;
+DROP TABLE t1;
+
+--echo # Table "t1" in the schema "public"
+CREATE TABLE t1 ENGINE=CONNECT TABNAME='public.t1' TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr';
+SHOW CREATE TABLE t1;
+SELECT * FROM t1;
+DROP TABLE t1;
+
+--echo # Table "t1" in the schema "schema1"
+CREATE TABLE t1 ENGINE=CONNECT TABNAME='schema1.t1' CHARSET=utf8 DATA_CHARSET=utf8 TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr';
+SHOW CREATE TABLE t1;
+SELECT * FROM t1;
+
+CREATE TABLE t2 AS SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+SELECT * FROM t2;
+DROP TABLE t2;
+
+CREATE VIEW v1 AS SELECT * FROM t1;
+SELECT * FROM v1;
+DROP VIEW v1;
+DROP TABLE t1;
+
+--echo # View "v1" in the schema "schema1"
+CREATE TABLE t1 ENGINE=CONNECT TABNAME='schema1.v1' CHARSET=utf8 DATA_CHARSET=utf8 TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr';
+SHOW CREATE TABLE t1;
+SELECT * FROM t1;
+
+CREATE TABLE t2 AS SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+SELECT * FROM t2;
+DROP TABLE t2;
+
+CREATE VIEW v1 AS SELECT * FROM t1;
+SELECT * FROM v1;
+DROP VIEW v1;
+DROP TABLE t1;
+
+--echo # Table "t2" in the schema "schema1"
+CREATE TABLE t1 ENGINE=CONNECT TABNAME='schema1.t2' CHARSET=utf8 DATA_CHARSET=utf8 TABLE_TYPE=ODBC CONNECTION='DSN=ConnectEnginePostgresql;UID=mtr;PWD=mtr';
+SHOW CREATE TABLE t1;
+SELECT * FROM t1;
+
+CREATE TABLE t2 AS SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+SELECT * FROM t2;
+DROP TABLE t2;
+
+CREATE VIEW v1 AS SELECT * FROM t1;
+SELECT * FROM v1;
+DROP VIEW v1;
+DROP TABLE t1;

storage/connect/odbconn.cpp

@@ -1986,6 +1986,87 @@ bool ODBConn::GetDrivers(PQRYRES qrp)
   return rv;
   } // end of GetDrivers
 
+
+/**
+  A helper class to split an optionally qualified table name into components.
+  These formats are understood:
+    "CatalogName.SchemaName.TableName"
+    "SchemaName.TableName"
+    "TableName"
+*/
+class SQLQualifiedName
+{
+  static const uint max_parts= 3; /* Catalog.Schema.Table */
+  MYSQL_LEX_STRING m_part[max_parts];
+  char m_buf[512];
+  void lex_string_set(MYSQL_LEX_STRING *S, char *str, size_t length)
+  {
+    S->str= str;
+    S->length= length;
+  }
+  void lex_string_shorten_down(MYSQL_LEX_STRING *S, size_t offs)
+  {
+    DBUG_ASSERT(offs <= S->length);
+    S->str+= offs;
+    S->length-= offs;
+  }
+  /*
+    Find the rightmost '.' delimiter and return the length
+    of the qualifier, including the rightmost '.' delimier.
+    For example, for the string {"a.b.c",5} it will return 4,
+    which is the length of the qualifier "a.b."
+  */
+  size_t lex_string_find_qualifier(MYSQL_LEX_STRING *S)
+  {
+    size_t i;
+    for (i= S->length; i > 0; i--)
+    {
+      if (S->str[i - 1] == '.')
+      {
+        S->str[i - 1]= '\0';
+        return i;
+      }
+    }
+    return 0;
+  }
+public:
+  /*
+    Initialize to the given optionally qualified name.
+    NULL pointer in "name" is supported.
+  */
+  SQLQualifiedName(const char *name)
+  {
+    size_t len, i= 0;
+    if (!name)
+      goto ret;
+    /* Initialize the first (rightmost) part */
+    lex_string_set(&m_part[0], m_buf,
+                   strmake(m_buf, name, sizeof(m_buf) - 1) - m_buf);
+    /* Initialize the other parts, if exist. */
+    for (i= 1; i < max_parts; i++)
+    {
+      if (!(len= lex_string_find_qualifier(&m_part[i - 1])))
+        break;
+      lex_string_set(&m_part[i], m_part[i - 1].str, len - 1);
+      lex_string_shorten_down(&m_part[i - 1], len);
+    }
+ret:
+    /* Initialize the remaining parts */
+    for ( ; i < max_parts; i++)
+      lex_string_set(&m_part[i], NULL, 0);
+  }
+  SQLCHAR *ptr(uint i)
+  {
+    DBUG_ASSERT(i < max_parts);
+    return (SQLCHAR *) (m_part[i].length ? m_part[i].str : NULL);
+  }
+  size_t length(uint i)
+  {
+    DBUG_ASSERT(i < max_parts);
+    return m_part[i].length;
+  }
+};
+
 /***********************************************************************/
 /*  Allocate recset and call SQLTables, SQLColumns or SQLPrimaryKeys.  */
 /***********************************************************************/
@@ -2048,29 +2129,38 @@ int ODBConn::GetCatInfo(CATPARM *cap)
     } else
       ThrowDBX("0-sized result");
 
+    SQLQualifiedName name((const char *) cap->Tab);
     // Now do call the proper ODBC API
     switch (cap->Id) {
       case CAT_TAB:
 //      rc = SQLSetStmtAttr(hstmt, SQL_ATTR_METADATA_ID,
 //                                (SQLPOINTER)false, 0);
         fnc = "SQLTables";
-        rc = SQLTables(hstmt, NULL, 0, NULL, 0, cap->Tab, SQL_NTS,
-                                                cap->Pat, SQL_NTS);
+        rc = SQLTables(hstmt, name.ptr(2), name.length(2),
+                              name.ptr(1), name.length(1),
+                              name.ptr(0), name.length(0),
+                              cap->Pat, SQL_NTS);
         break;
       case CAT_COL:
 //      rc = SQLSetStmtAttr(hstmt, SQL_ATTR_METADATA_ID,
 //                                (SQLPOINTER)true, 0);
         fnc = "SQLColumns";
-        rc = SQLColumns(hstmt, NULL, 0, NULL, 0, cap->Tab, SQL_NTS,
-                                                 cap->Pat, SQL_NTS);
+        rc = SQLColumns(hstmt, name.ptr(2), name.length(2),
+                               name.ptr(1), name.length(1),
+                               name.ptr(0), name.length(0),
+                               cap->Pat, SQL_NTS);
         break;
       case CAT_KEY:
         fnc = "SQLPrimaryKeys";
-        rc = SQLPrimaryKeys(hstmt, NULL, 0, NULL, 0, cap->Tab, SQL_NTS);
+        rc = SQLPrimaryKeys(hstmt, name.ptr(2), name.length(2),
+                                   name.ptr(1), name.length(1),
+                                   name.ptr(0), name.length(0));
         break;
       case CAT_STAT:
         fnc = "SQLStatistics";
-        rc = SQLStatistics(hstmt, NULL, 0, NULL, 0, cap->Tab, SQL_NTS,
+        rc = SQLStatistics(hstmt, name.ptr(2), name.length(2),
+                                  name.ptr(1), name.length(1),
+                                  name.ptr(0), name.length(0),
                                   cap->Unique, cap->Accuracy);
         break;
       case CAT_SPC: