IGNITE-27207 Thin client: SqlFieldsQuery initiator ID support - Fixes #12550.

Signed-off-by: Aleksey Plekhanov <[email protected]>

Author: alex-plekhanov

modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientUtils.java

@@ -62,6 +62,8 @@
 import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.internal.client.thin.ProtocolBitmaskFeature.CACHE_STORAGES;
+import static org.apache.ignite.internal.client.thin.ProtocolBitmaskFeature.QRY_INITIATOR_ID;
+import static org.apache.ignite.internal.client.thin.ProtocolBitmaskFeature.QRY_PARTITIONS_BATCH_SIZE;
 import static org.apache.ignite.internal.client.thin.ProtocolVersionFeature.EXPIRY_POLICY;
 import static org.apache.ignite.internal.client.thin.ProtocolVersionFeature.QUERY_ENTITY_PRECISION_AND_SCALE;
 import static org.apache.ignite.internal.processors.platform.cache.expiry.PlatformExpiryPolicy.convertDuration;
@@ -519,7 +521,7 @@ ClientCacheConfiguration cacheConfiguration(BinaryInputStream in, ProtocolContex
     }
 
     /** Serialize SQL field query to stream. */
-    void write(SqlFieldsQuery qry, BinaryOutputStream out) {
+    void write(SqlFieldsQuery qry, BinaryOutputStream out, ProtocolContext protocolCtx) {
         writeObject(out, qry.getSchema());
         out.writeInt(qry.getPageSize());
         out.writeInt(-1); // do not limit
@@ -535,16 +537,21 @@ void write(SqlFieldsQuery qry, BinaryOutputStream out) {
         out.writeLong(qry.getTimeout());
         out.writeBoolean(true); // include column names
 
-        if (qry.getPartitions() != null) {
-            out.writeInt(qry.getPartitions().length);
+        if (protocolCtx.isFeatureSupported(QRY_PARTITIONS_BATCH_SIZE)) {
+            if (qry.getPartitions() != null) {
+                out.writeInt(qry.getPartitions().length);
 
-            for (int part : qry.getPartitions())
-                out.writeInt(part);
+                for (int part : qry.getPartitions())
+                    out.writeInt(part);
+            }
+            else
+                out.writeInt(-1);
+
+            out.writeInt(qry.getUpdateBatchSize());
         }
-        else
-            out.writeInt(-1);
 
-        out.writeInt(qry.getUpdateBatchSize());
+        if (protocolCtx.isFeatureSupported(QRY_INITIATOR_ID))
+            writeObject(out, qry.getQueryInitiatorId());
     }
 
     /** Write Ignite binary object to output stream. */

modules/core/src/main/java/org/apache/ignite/internal/client/thin/ProtocolBitmaskFeature.java

@@ -112,7 +112,10 @@ public enum ProtocolBitmaskFeature {
     SQL_CACHE_CREATION(21),
 
     /** Data-center information. */
-    DC_AWARE(22);
+    DC_AWARE(22),
+
+    /** SqlFieldsQuery initiatorId property. */
+    QRY_INITIATOR_ID(23);
 
     /** */
     private static final EnumSet<ProtocolBitmaskFeature> ALL_FEATURES_AS_ENUM_SET =

modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpClientCache.java

@@ -1084,7 +1084,7 @@ else if (qry instanceof IndexQuery)
                     ? transactions.tx()
                     : null
             );
-            serDes.write(qry, payloadCh.out());
+            serDes.write(qry, payloadCh.out(), payloadCh.clientChannel().protocolCtx());
         };
 
         return new ClientFieldsQueryCursor<>(new ClientFieldsQueryPager(

modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpIgniteClient.java

@@ -372,7 +372,7 @@ public <K, V> ClientCache<K, V> getOrCreateCache(ClientCacheConfiguration cfg, b
             else
                 out.writeByte(flags);
 
-            serDes.write(qry, out);
+            serDes.write(qry, out, payloadCh.clientChannel().protocolCtx());
         };
 
         return new ClientFieldsQueryCursor<>(new ClientFieldsQueryPager(

modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientBitmaskFeature.java

@@ -109,7 +109,10 @@ public enum ClientBitmaskFeature implements ThinProtocolFeature {
     SQL_CACHE_CREATION(21),
 
     /** Data-center information. */
-    DC_AWARE(22);
+    DC_AWARE(22),
+
+    /** SqlFieldsQuery initiatorId property. */
+    QRY_INITIATOR_ID(23);
 
     /** */
     private static final EnumSet<ClientBitmaskFeature> ALL_FEATURES_AS_ENUM_SET =

modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/cache/ClientCacheSqlFieldsQueryRequest.java

@@ -53,6 +53,9 @@ public class ClientCacheSqlFieldsQueryRequest extends ClientCacheQueryRequest im
     /** Update batch size. */
     private final Integer updateBatchSize;
 
+    /** Query initiator ID. */
+    private final String initiatorId;
+
     /**
      * Ctor.
      *
@@ -120,6 +123,11 @@ public ClientCacheSqlFieldsQueryRequest(BinaryReaderEx reader,
             partitions = null;
             updateBatchSize = null;
         }
+
+        if (protocolCtx.isFeatureSupported(ClientBitmaskFeature.QRY_INITIATOR_ID))
+            initiatorId = reader.readString();
+        else
+            initiatorId = null;
     }
 
     /** {@inheritDoc} */
@@ -132,7 +140,7 @@ public ClientCacheSqlFieldsQueryRequest(BinaryReaderEx reader,
         ctx.incrementCursors();
 
         try {
-            qry.setQueryInitiatorId(ctx.clientDescriptor());
+            qry.setQueryInitiatorId(initiatorId == null ? ctx.clientDescriptor() : initiatorId);
 
             // If cacheId is provided, we must check the cache for existence.
             if (cacheId() != 0) {

modules/indexing/src/test/java/org/apache/ignite/client/FunctionalQueryTest.java

@@ -317,6 +317,30 @@ public void testEmptyQuery() {
         }
     }
 
+    /** Tests {@link SqlFieldsQuery} initiator ID parameter. */
+    @Test
+    public void testQueryInitiatorId() {
+        try (Ignite ignored = Ignition.start(Config.getServerConfiguration());
+             IgniteClient client = Ignition.startClient(new ClientConfiguration().setAddresses(Config.SERVER))
+        ) {
+            String initiatorId = "test";
+
+            SqlFieldsQuery qry = new SqlFieldsQuery("SELECT INITIATOR_ID FROM SYS.SQL_QUERIES").setQueryInitiatorId(initiatorId);
+
+            List<List<?>> res = client.query(qry).getAll();
+
+            assertEquals(1, res.size());
+            assertEquals(initiatorId, res.get(0).get(0));
+
+            ClientCache<Object, Object> cache = client.getOrCreateCache(Config.DEFAULT_CACHE_NAME);
+
+            res = cache.query(qry).getAll();
+
+            assertEquals(1, res.size());
+            assertEquals(initiatorId, res.get(0).get(0));
+        }
+    }
+
     /** */
     private static ClientConfiguration getClientConfiguration() {
         return new ClientConfiguration().setAddresses(Config.SERVER)