SQL: Fix qualified attribute resolution with CCS (elastic#81320) (ela…

…stic#81570)

In case the fields are qualified only by the index name in a
cross cluster search query, the field resolution was failing since an
attribute's qualification contains the cluster name, while a field's doesn't.

This changes the existing  `Attribute#qualifiedName()` to return a qualification
including the index name only and adds (for completion) a new method,
`fullyQualifiedName()`, which will return the 'cluster:index' qualification.

This will allow queries like this to work:
 `SELECT remote_index.field FROM remote_cluster:remote_index`
or
 `SELECT remote_index.field FROM remote_index` when the cluster is provided
through the xDBC "cluster" parameter.

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/expression/Attribute.java

@@ -6,13 +6,15 @@
  */
 package org.elasticsearch.xpack.ql.expression;
 
+import org.elasticsearch.core.Tuple;
 import org.elasticsearch.xpack.ql.tree.Source;
 import org.elasticsearch.xpack.ql.type.DataType;
 
 import java.util.List;
 import java.util.Objects;
 
 import static java.util.Collections.emptyList;
+import static org.elasticsearch.xpack.ql.util.StringUtils.splitQualifiedIndex;
 
 /**
  * {@link Expression}s that can be materialized and describe properties of the derived table.
@@ -31,6 +33,8 @@ public abstract class Attribute extends NamedExpression {
     // empty - such as a top level attribute in SELECT cause
     // present - table name or a table name alias
     private final String qualifier;
+    // cluster name in the qualifier (if any)
+    private final String cluster;
 
     // can the attr be null - typically used in JOINs
     private final Nullability nullability;
@@ -45,7 +49,14 @@ public Attribute(Source source, String name, String qualifier, Nullability nulla
 
     public Attribute(Source source, String name, String qualifier, Nullability nullability, NameId id, boolean synthetic) {
         super(source, name, emptyList(), id, synthetic);
-        this.qualifier = qualifier;
+        if (qualifier != null) {
+            Tuple<String, String> splitQualifier = splitQualifiedIndex(qualifier);
+            this.cluster = splitQualifier.v1();
+            this.qualifier = splitQualifier.v2();
+        } else {
+            this.cluster = null;
+            this.qualifier = null;
+        }
         this.nullability = nullability;
     }
 

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/index/IndexResolver.java

@@ -73,8 +73,8 @@
 import static org.elasticsearch.xpack.ql.type.DataTypes.OBJECT;
 import static org.elasticsearch.xpack.ql.type.DataTypes.TEXT;
 import static org.elasticsearch.xpack.ql.type.DataTypes.UNSUPPORTED;
-import static org.elasticsearch.xpack.ql.util.RemoteClusterUtils.qualifyAndJoinIndices;
-import static org.elasticsearch.xpack.ql.util.RemoteClusterUtils.splitQualifiedIndex;
+import static org.elasticsearch.xpack.ql.util.StringUtils.qualifyAndJoinIndices;
+import static org.elasticsearch.xpack.ql.util.StringUtils.splitQualifiedIndex;
 
 public class IndexResolver {
 

x-pack/plugin/ql/src/main/java/org/elasticsearch/xpack/ql/util/StringUtils.java

@@ -21,8 +21,11 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Locale;
+import java.util.StringJoiner;
 
 import static java.util.stream.Collectors.toList;
+import static org.elasticsearch.transport.RemoteClusterAware.REMOTE_CLUSTER_INDEX_SEPARATOR;
+import static org.elasticsearch.transport.RemoteClusterAware.buildRemoteIndexName;
 
 public final class StringUtils {
 
@@ -334,4 +337,23 @@ public static String ordinal(int i) {
 
         }
     }
+
+    public static Tuple<String, String> splitQualifiedIndex(String indexName) {
+        int separatorOffset = indexName.indexOf(REMOTE_CLUSTER_INDEX_SEPARATOR);
+        return separatorOffset > 0
+            ? Tuple.tuple(indexName.substring(0, separatorOffset), indexName.substring(separatorOffset + 1))
+            : Tuple.tuple(null, indexName);
+    }
+
+    public static String qualifyAndJoinIndices(String cluster, String[] indices) {
+        StringJoiner sj = new StringJoiner(",");
+        for (String index : indices) {
+            sj.add(cluster != null ? buildRemoteIndexName(cluster, index) : index);
+        }
+        return sj.toString();
+    }
+
+    public static boolean isQualified(String indexWildcard) {
+        return indexWildcard.indexOf(REMOTE_CLUSTER_INDEX_SEPARATOR) > 0;
+    }
 }

x-pack/plugin/ql/src/test/java/org/elasticsearch/xpack/ql/util/RemoteClusterUtilsTests.java

@@ -10,9 +10,9 @@
 import org.elasticsearch.core.Tuple;
 import org.elasticsearch.test.ESTestCase;
 
-import static org.elasticsearch.xpack.ql.util.RemoteClusterUtils.isQualified;
-import static org.elasticsearch.xpack.ql.util.RemoteClusterUtils.qualifyAndJoinIndices;
-import static org.elasticsearch.xpack.ql.util.RemoteClusterUtils.splitQualifiedIndex;
+import static org.elasticsearch.xpack.ql.util.StringUtils.isQualified;
+import static org.elasticsearch.xpack.ql.util.StringUtils.qualifyAndJoinIndices;
+import static org.elasticsearch.xpack.ql.util.StringUtils.splitQualifiedIndex;
 
 public class RemoteClusterUtilsTests extends ESTestCase {
     public void testSplitQualifiedIndex() {

x-pack/plugin/sql/qa/server/multi-cluster-with-security/src/test/java/org/elasticsearch/xpack/sql/qa/multi_cluster_with_security/JdbcCatalogIT.java

@@ -70,6 +70,30 @@ public void testQueryCatalogPrecedence() throws Exception {
         }
     }
 
+    public void testQueryWithQualifierAndSetCatalog() throws Exception {
+        try (Connection es = esJdbc()) {
+            PreparedStatement ps = es.prepareStatement("SELECT " + INDEX_NAME + ".zero FROM " + INDEX_NAME);
+            es.setCatalog(REMOTE_CLUSTER_NAME);
+            ResultSet rs = ps.executeQuery();
+            assertTrue(rs.next());
+            assertEquals(0, rs.getInt(1));
+            assertFalse(rs.next());
+        }
+    }
+
+    public void testQueryWithQualifiedFieldAndIndex() throws Exception {
+        try (Connection es = esJdbc()) {
+            PreparedStatement ps = es.prepareStatement(
+                "SELECT " + INDEX_NAME + ".zero FROM " + buildRemoteIndexName(REMOTE_CLUSTER_NAME, INDEX_NAME)
+            );
+            es.setCatalog(LOCAL_CLUSTER_NAME); // set, but should be ignored
+            ResultSet rs = ps.executeQuery();
+            assertTrue(rs.next());
+            assertEquals(0, rs.getInt(1));
+            assertFalse(rs.next());
+        }
+    }
+
     public void testCatalogDependentCommands() throws Exception {
         for (String query : List.of(
             "SHOW TABLES \"" + INDEX_NAME + "\"",

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/plan/logical/command/sys/SysColumns.java

@@ -41,8 +41,8 @@
 import static org.elasticsearch.xpack.ql.type.DataTypes.SHORT;
 import static org.elasticsearch.xpack.ql.type.DataTypes.isPrimitive;
 import static org.elasticsearch.xpack.ql.type.DataTypes.isString;
-import static org.elasticsearch.xpack.ql.util.RemoteClusterUtils.isQualified;
-import static org.elasticsearch.xpack.ql.util.RemoteClusterUtils.splitQualifiedIndex;
+import static org.elasticsearch.xpack.ql.util.StringUtils.isQualified;
+import static org.elasticsearch.xpack.ql.util.StringUtils.splitQualifiedIndex;
 import static org.elasticsearch.xpack.sql.type.SqlDataTypes.displaySize;
 import static org.elasticsearch.xpack.sql.type.SqlDataTypes.metaSqlDataType;
 import static org.elasticsearch.xpack.sql.type.SqlDataTypes.metaSqlDateTimeSub;