Fix escaping in metadata calls to SQL Server driver

plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/BaseJdbcClient.java

@@ -345,8 +345,8 @@ protected ResultSet getColumns(JdbcTableHandle tableHandle, DatabaseMetaData met
         RemoteTableName remoteTableName = tableHandle.getRequiredNamedRelation().getRemoteTableName();
         return metadata.getColumns(
                 remoteTableName.getCatalogName().orElse(null),
-                escapeNamePattern(remoteTableName.getSchemaName(), metadata.getSearchStringEscape()).orElse(null),
-                escapeNamePattern(remoteTableName.getTableName(), metadata.getSearchStringEscape()),
+                escapeObjectNameForMetadataQuery(remoteTableName.getSchemaName(), metadata.getSearchStringEscape()).orElse(null),
+                escapeObjectNameForMetadataQuery(remoteTableName.getTableName(), metadata.getSearchStringEscape()),
                 null);
     }
 
@@ -875,8 +875,8 @@ public ResultSet getTables(Connection connection, Optional<String> remoteSchemaN
         DatabaseMetaData metadata = connection.getMetaData();
         return metadata.getTables(
                 connection.getCatalog(),
-                escapeNamePattern(remoteSchemaName, metadata.getSearchStringEscape()).orElse(null),
-                escapeNamePattern(remoteTableName, metadata.getSearchStringEscape()).orElse(null),
+                escapeObjectNameForMetadataQuery(remoteSchemaName, metadata.getSearchStringEscape()).orElse(null),
+                escapeObjectNameForMetadataQuery(remoteTableName, metadata.getSearchStringEscape()).orElse(null),
                 getTableTypes().map(types -> types.toArray(String[]::new)).orElse(null));
     }
 
@@ -1137,12 +1137,12 @@ public static String varcharLiteral(String value)
         return "'" + value.replace("'", "''") + "'";
     }
 
-    protected static Optional<String> escapeNamePattern(Optional<String> name, String escape)
+    protected Optional<String> escapeObjectNameForMetadataQuery(Optional<String> name, String escape)
     {
-        return name.map(string -> escapeNamePattern(string, escape));
+        return name.map(string -> escapeObjectNameForMetadataQuery(string, escape));
     }
 
-    private static String escapeNamePattern(String name, String escape)
+    protected String escapeObjectNameForMetadataQuery(String name, String escape)
     {
         requireNonNull(name, "name is null");
         requireNonNull(escape, "escape is null");

plugin/trino-mariadb/src/main/java/io/trino/plugin/mariadb/MariaDbClient.java

@@ -246,7 +246,7 @@ public ResultSet getTables(Connection connection, Optional<String> schemaName, O
         return metadata.getTables(
                 schemaName.orElse(null),
                 null,
-                escapeNamePattern(tableName, metadata.getSearchStringEscape()).orElse(null),
+                escapeObjectNameForMetadataQuery(tableName, metadata.getSearchStringEscape()).orElse(null),
                 getTableTypes().map(types -> types.toArray(String[]::new)).orElse(null));
     }
 

plugin/trino-mysql/src/main/java/io/trino/plugin/mysql/MySqlClient.java

@@ -313,7 +313,7 @@ public ResultSet getTables(Connection connection, Optional<String> schemaName, O
         return metadata.getTables(
                 schemaName.orElse(null),
                 null,
-                escapeNamePattern(tableName, metadata.getSearchStringEscape()).orElse(null),
+                escapeObjectNameForMetadataQuery(tableName, metadata.getSearchStringEscape()).orElse(null),
                 getTableTypes().map(types -> types.toArray(String[]::new)).orElse(null));
     }
 

plugin/trino-singlestore/src/main/java/io/trino/plugin/singlestore/SingleStoreClient.java

@@ -316,7 +316,7 @@ public ResultSet getTables(Connection connection, Optional<String> schemaName, O
         return metadata.getTables(
                 schemaName.orElse(null),
                 null,
-                escapeNamePattern(tableName, metadata.getSearchStringEscape()).orElse(null),
+                escapeObjectNameForMetadataQuery(tableName, metadata.getSearchStringEscape()).orElse(null),
                 getTableTypes().map(types -> types.toArray(String[]::new)).orElse(null));
     }
 

plugin/trino-sqlserver/src/main/java/io/trino/plugin/sqlserver/SqlServerClient.java

@@ -675,6 +675,26 @@ private static Map<String, String> getColumnNameToStatisticsName(JdbcTableHandle
         return columnNameToStatisticsName;
     }
 
+    // SQL Server has non-standard LIKE semantics:
+    // https://learn.microsoft.com/en-us/sql/t-sql/language-elements/like-transact-sql?redirectedfrom=MSDN&view=sql-server-ver16#arguments
+    // and apparently this applies to DatabaseMetaData calls too.@Override
+    @Override
+    protected String escapeObjectNameForMetadataQuery(String name, String escape)
+    {
+        requireNonNull(name, "name is null");
+        requireNonNull(escape, "escape is null");
+        checkArgument(!escape.isEmpty(), "Escape string must not be empty");
+        checkArgument(!escape.equals("_"), "Escape string must not be '_'");
+        checkArgument(!escape.equals("%"), "Escape string must not be '%'");
+        name = name.replace(escape, escape + escape);
+        name = name.replace("_", escape + "_");
+        name = name.replace("%", escape + "%");
+        // SQLServer also treats [ and ] as wildcard characters
+        name = name.replace("]", escape + "]");
+        name = name.replace("[", escape + "[");
+        return name;
+    }
+
     @Override
     public Optional<PreparedQuery> implementJoin(
             ConnectorSession session,

plugin/trino-sqlserver/src/test/java/io/trino/plugin/sqlserver/TestSqlServerConnectorTest.java

@@ -35,11 +35,15 @@
 import static io.trino.plugin.sqlserver.SqlServerSessionProperties.BULK_COPY_FOR_WRITE;
 import static io.trino.plugin.sqlserver.SqlServerSessionProperties.BULK_COPY_FOR_WRITE_LOCK_DESTINATION_TABLE;
 import static io.trino.testing.DataProviders.cartesianProduct;
+import static io.trino.testing.DataProviders.toDataProvider;
 import static io.trino.testing.DataProviders.trueFalse;
 import static io.trino.testing.sql.TestTable.randomTableSuffix;
 import static java.lang.String.format;
+import static java.util.Locale.ENGLISH;
 import static java.util.stream.Collectors.joining;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
 
 public class TestSqlServerConnectorTest
         extends BaseSqlServerConnectorTest
@@ -151,6 +155,21 @@ public void testInsertWriteBulkinessWithTimestamps(String timestampType)
         }
     }
 
+    // TODO move test to BaseConnectorTest
+    @Test(dataProvider = "testTableNameDataProvider")
+    public void testCreateAndDropTableWithSpecialCharacterName(String tableName)
+    {
+        String tableNameInSql = "\"" + tableName.replace("\"", "\"\"") + "\"";
+        // Until https://github.com/trinodb/trino/issues/17 the table name is effectively lowercase
+        tableName = tableName.toLowerCase(ENGLISH);
+        assertUpdate("CREATE TABLE " + tableNameInSql + " (a bigint, b double, c varchar(50))");
+        assertTrue(getQueryRunner().tableExists(getSession(), tableName));
+        assertTableColumnNames(tableNameInSql, "a", "b", "c");
+
+        assertUpdate("DROP TABLE " + tableNameInSql);
+        assertFalse(getQueryRunner().tableExists(getSession(), tableName));
+    }
+
     private int getTableOperationsCount(String operation, String table)
             throws SQLException
     {
@@ -195,4 +214,44 @@ public static Object[][] timestampTypes()
                 {"timestamp(12)"}
         };
     }
+
+    // TODO replace TableNameDataProvider and ColumnNameDataProvider with ObjectNameDataProvider
+    //  to one big single list of all special character cases, current list has additional special bracket cases,
+    //  please don't forget to use this list as base
+    @DataProvider
+    public Object[][] testTableNameDataProvider()
+    {
+        return testTableNameTestData().stream()
+                .collect(toDataProvider());
+    }
+
+    private List<String> testTableNameTestData()
+    {
+        return ImmutableList.<String>builder()
+                .add("lowercase")
+                .add("UPPERCASE")
+                .add("MixedCase")
+                .add("an_underscore")
+                .add("a-hyphen-minus") // ASCII '-' is HYPHEN-MINUS in Unicode
+                .add("a space")
+                .add("atrailingspace ")
+                .add(" aleadingspace")
+                .add("a.dot")
+                .add("a,comma")
+                .add("a:colon")
+                .add("a;semicolon")
+                .add("an@at")
+                .add("a\"quote")
+                .add("an'apostrophe")
+                .add("a`backtick`")
+                .add("a/slash`")
+                .add("a\\backslash`")
+                .add("adigit0")
+                .add("0startwithdigit")
+                .add("[brackets]")
+                .add("brackets[]inside")
+                .add("open[bracket")
+                .add("close]bracket")
+                .build();
+    }
 }