Fix qualified attribute resolution with CCS

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,16 @@
  */
 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.transport.RemoteClusterAware.buildRemoteIndexName;
+import static org.elasticsearch.xpack.ql.util.RemoteClusterUtils.splitQualifiedIndex;
 
 /**
  * {@link Expression}s that can be materialized and describe properties of the derived table.
@@ -31,6 +34,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 +50,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;
     }
 
@@ -62,6 +74,10 @@ public String qualifiedName() {
         return qualifier == null ? name() : qualifier + "." + name();
     }
 
+    public String fullyQualifiedName() {
+        return qualifier == null ? name() : buildRemoteIndexName(cluster, qualifier) + "." + name();
+    }
+
     @Override
     public Nullability nullable() {
         return nullability;

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 + "\"",