SQL: Fix catalog filtering in SYS COLUMNS (#40583)

Properly treat '%' as a wildcard for catalog filtering instead of doing
a straight string match.
Table filtering now considers aliases as well.
Add escaping char for LIKE queries with user defined params
Fix monotony of ORDINAL_POSITION
Add integration test for SYS COLUMNS - currently running only inside
single_node since the cluster name is test dependent.
Add pattern unescaping for index names

Fix #40582

(cherry picked from commit 8e61b77)

x-pack/plugin/sql/jdbc/src/main/java/org/elasticsearch/xpack/sql/jdbc/JdbcDatabaseMetaData.java

@@ -33,6 +33,8 @@
  */
 class JdbcDatabaseMetaData implements DatabaseMetaData, JdbcWrapper {
 
+    private static final String WILDCARD = "%";
+
     private final JdbcConnection con;
 
     JdbcDatabaseMetaData(JdbcConnection con) {
@@ -713,19 +715,19 @@ private boolean isDefaultCatalog(String catalog) throws SQLException {
         // null means catalog info is irrelevant
         // % means return all catalogs
         // EMPTY means return those without a catalog
-        return catalog == null || catalog.equals(EMPTY) || catalog.equals("%") || catalog.equals(defaultCatalog());
+        return catalog == null || catalog.equals(EMPTY) || catalog.equals(WILDCARD) || catalog.equals(defaultCatalog());
     }
 
     private boolean isDefaultSchema(String schema) {
         // null means schema info is irrelevant
         // % means return all schemas`
         // EMPTY means return those without a schema
-        return schema == null || schema.equals(EMPTY) || schema.equals("%");
+        return schema == null || schema.equals(EMPTY) || schema.equals(WILDCARD);
     }
 
     @Override
     public ResultSet getTables(String catalog, String schemaPattern, String tableNamePattern, String[] types) throws SQLException {
-        String statement = "SYS TABLES CATALOG LIKE ? LIKE ?";
+        String statement = "SYS TABLES CATALOG LIKE ? ESCAPE '\\' LIKE ? ESCAPE '\\' ";
 
         if (types != null && types.length > 0) {
             statement += " TYPE ?";
@@ -738,8 +740,8 @@ public ResultSet getTables(String catalog, String schemaPattern, String tableNam
         }
 
         PreparedStatement ps = con.prepareStatement(statement);
-        ps.setString(1, catalog != null ? catalog.trim() : "%");
-        ps.setString(2, tableNamePattern != null ? tableNamePattern.trim() : "%");
+        ps.setString(1, catalog != null ? catalog.trim() : WILDCARD);
+        ps.setString(2, tableNamePattern != null ? tableNamePattern.trim() : WILDCARD);
 
         if (types != null && types.length > 0) {
             for (int i = 0; i < types.length; i++) {
@@ -784,14 +786,15 @@ public ResultSet getTableTypes() throws SQLException {
         return memorySet(con.cfg, columnInfo("TABLE_TYPES", "TABLE_TYPE"), data);
     }
 
+
     @Override
     public ResultSet getColumns(String catalog, String schemaPattern, String tableNamePattern, String columnNamePattern)
             throws SQLException {
-        PreparedStatement ps = con.prepareStatement("SYS COLUMNS CATALOG ? TABLE LIKE ? LIKE ?");
-        // TODO: until passing null works, pass an empty string
-        ps.setString(1, catalog != null ? catalog.trim() : EMPTY);
-        ps.setString(2, tableNamePattern != null ? tableNamePattern.trim() : "%");
-        ps.setString(3, columnNamePattern != null ? columnNamePattern.trim() : "%");
+        PreparedStatement ps = con.prepareStatement("SYS COLUMNS CATALOG ? TABLE LIKE ? ESCAPE '\\' LIKE ? ESCAPE '\\'");
+        // NB: catalog is not a pattern hence why null is send instead
+        ps.setString(1, catalog != null ? catalog.trim() : null);
+        ps.setString(2, tableNamePattern != null ? tableNamePattern.trim() : WILDCARD);
+        ps.setString(3, columnNamePattern != null ? columnNamePattern.trim() : WILDCARD);
         return ps.executeQuery();
     }
 

x-pack/plugin/sql/qa/security/src/test/java/org/elasticsearch/xpack/sql/qa/security/JdbcSecurityIT.java

@@ -232,13 +232,13 @@ public void expectUnknownColumn(String user, String sql, String column) throws E
         public void checkNoMonitorMain(String user) throws Exception {
             // Without monitor/main the JDBC driver - ES server version comparison doesn't take place, which fails everything else
             expectUnauthorized("cluster:monitor/main", user, () -> es(userProperties(user)));
-            expectUnauthorized("cluster:monitor/main", user, () -> es(userProperties(user)).getMetaData().getDatabaseMajorVersion()); 
+            expectUnauthorized("cluster:monitor/main", user, () -> es(userProperties(user)).getMetaData().getDatabaseMajorVersion());
             expectUnauthorized("cluster:monitor/main", user, () -> es(userProperties(user)).getMetaData().getDatabaseMinorVersion());
-            expectUnauthorized("cluster:monitor/main", user, 
+            expectUnauthorized("cluster:monitor/main", user,
                     () -> es(userProperties(user)).createStatement().executeQuery("SELECT * FROM test"));
-            expectUnauthorized("cluster:monitor/main", user, 
+            expectUnauthorized("cluster:monitor/main", user,
                     () -> es(userProperties(user)).createStatement().executeQuery("SHOW TABLES LIKE 'test'"));
-            expectUnauthorized("cluster:monitor/main", user, 
+            expectUnauthorized("cluster:monitor/main", user,
                     () -> es(userProperties(user)).createStatement().executeQuery("DESCRIBE test"));
         }
 
@@ -292,7 +292,7 @@ public void testMetaDataGetColumnsWorksAsFullAccess() throws Exception {
         expectActionMatchesAdmin(
                 con -> con.getMetaData().getColumns(null, "%", "%t", "%"),
             "full_access",
-                con -> con.getMetaData().getColumns(null, "%", "%", "%"));
+                con -> con.getMetaData().getColumns(null, "%", "%t", "%"));
     }
 
     public void testMetaDataGetColumnsWithNoAccess() throws Exception {

x-pack/plugin/sql/qa/single-node/src/test/java/org/elasticsearch/xpack/sql/qa/single_node/JdbcCsvSpecIT.java

@@ -5,10 +5,26 @@
  */
 package org.elasticsearch.xpack.sql.qa.single_node;
 
+import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
+
 import org.elasticsearch.xpack.sql.qa.jdbc.CsvSpecTestCase;
 import org.elasticsearch.xpack.sql.qa.jdbc.CsvTestUtils.CsvTestCase;
 
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.elasticsearch.xpack.sql.qa.jdbc.CsvTestUtils.specParser;
+
 public class JdbcCsvSpecIT extends CsvSpecTestCase {
+
+
+    @ParametersFactory(argumentFormatting = PARAM_FORMATTING)
+    public static List<Object[]> readScriptSpec() throws Exception {
+        List<Object[]> list = new ArrayList<>();
+        list.addAll(CsvSpecTestCase.readScriptSpec());
+        return readScriptSpec("/single-node-only/command-sys.csv-spec", specParser());
+    }
+
     public JdbcCsvSpecIT(String fileName, String groupName, String testName, Integer lineNumber, CsvTestCase testCase) {
         super(fileName, groupName, testName, lineNumber, testCase);
     }

x-pack/plugin/sql/qa/src/main/java/org/elasticsearch/xpack/sql/qa/jdbc/CsvTestUtils.java

@@ -155,6 +155,8 @@ private static String resolveColumnType(String type) {
                 return "timestamp";
             case "bt":
                 return "byte";
+            case "sh":
+                return "short";
             default:
                 return type;
         }

x-pack/plugin/sql/qa/src/main/java/org/elasticsearch/xpack/sql/qa/jdbc/JdbcTestUtils.java

@@ -212,7 +212,8 @@ public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IO
     @SuppressForbidden(reason = "need to open jar")
     private static JarInputStream getJarStream(URL resource) throws IOException {
         URLConnection con = resource.openConnection();
-        con.setDefaultUseCaches(false);
+        // do not to cache files (to avoid keeping file handles around)
+        con.setUseCaches(false);
         return new JarInputStream(con.getInputStream());
     }
 

x-pack/plugin/sql/qa/src/main/java/org/elasticsearch/xpack/sql/qa/jdbc/SpecBaseIntegrationTestCase.java

@@ -19,6 +19,7 @@
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.net.URL;
+import java.net.URLConnection;
 import java.nio.charset.StandardCharsets;
 import java.sql.Connection;
 import java.sql.ResultSet;
@@ -216,6 +217,9 @@ public interface Parser {
 
     @SuppressForbidden(reason = "test reads from jar")
     public static InputStream readFromJarUrl(URL source) throws IOException {
-        return source.openStream();
+        URLConnection con = source.openConnection();
+        // do not to cache files (to avoid keeping file handles around)
+        con.setUseCaches(false);
+        return con.getInputStream();
     }
 }

x-pack/plugin/sql/qa/src/main/resources/setup_mock_metadata_get_columns.sql

@@ -30,7 +30,7 @@ FROM DUAL
 UNION ALL
 SELECT null, 'test1', 'name.keyword', 12, 'KEYWORD', 32766, 2147483647, null, null,
   1, -- columnNullable
-  null, null, 12, 0, 2147483647, 1, 'YES', null, null, null, null, 'NO', 'NO'
+  null, null, 12, 0, 2147483647, 2, 'YES', null, null, null, null, 'NO', 'NO'
 FROM DUAL
 UNION ALL
 SELECT null, 'test2', 'date', 93, 'DATETIME', 29, 8, null, null,