Added sending statement parameters in multipart data request

Author: chernser

client-v2/src/main/java/com/clickhouse/client/api/internal/HttpAPIClientHelper.java

@@ -50,6 +50,8 @@
 import org.apache.hc.core5.http.impl.io.DefaultHttpResponseParserFactory;
 import org.apache.hc.core5.http.io.SocketConfig;
 import org.apache.hc.core5.http.io.entity.EntityTemplate;
+import org.apache.hc.client5.http.entity.mime.MultipartEntityBuilder;
+import java.io.ByteArrayOutputStream;
 import org.apache.hc.core5.http.protocol.HttpContext;
 import org.apache.hc.core5.io.CloseMode;
 import org.apache.hc.core5.io.IOCallback;
@@ -432,12 +434,45 @@ public ClassicHttpResponse executeRequest(Endpoint server, Map<String, Object> r
 //        req.setVersion(new ProtocolVersion("HTTP", 1, 0)); // to disable chunk transfer encoding
         addHeaders(req, requestConfig);
 
-
-        // setting entity. wrapping if compression is enabled
-        String contentEncoding = req.containsHeader(HttpHeaders.CONTENT_ENCODING) ? req.getHeader(HttpHeaders.CONTENT_ENCODING).getValue() : null;
-        req.setEntity(wrapRequestEntity(new EntityTemplate(-1, CONTENT_TYPE, contentEncoding , writeCallback),
-                lz4Factory,
-                requestConfig));
+        // Check if statement params exist - if so, send as multipart
+        boolean hasStatementParams = requestConfig.containsKey(KEY_STATEMENT_PARAMS);
+        Map<?, ?> statementParams = hasStatementParams ? 
+            (Map<?, ?>) requestConfig.get(KEY_STATEMENT_PARAMS) : null;
+        
+        HttpEntity requestEntity;
+        String contentEncoding = req.containsHeader(HttpHeaders.CONTENT_ENCODING) ? 
+            req.getHeader(HttpHeaders.CONTENT_ENCODING).getValue() : null;
+        
+        if (hasStatementParams && statementParams != null && !statementParams.isEmpty()) {
+            // Create multipart entity with query body and statement params
+            MultipartEntityBuilder multipartBuilder = MultipartEntityBuilder.create();
+            
+            // Add query/body data as binary body
+            if (writeCallback != null) {
+                // Write callback output to buffer, then add as binary body
+                ByteArrayOutputStream buffer = new ByteArrayOutputStream();
+                writeCallback.execute(buffer);
+                byte[] queryData = buffer.toByteArray();
+                multipartBuilder.addBinaryBody("query", queryData, CONTENT_TYPE, null);
+            }
+            
+            // Add statement params as multipart parts
+            for (Map.Entry<?, ?> entry : statementParams.entrySet()) {
+                String paramName = "param_" + entry.getKey().toString();
+                String paramValue = String.valueOf(entry.getValue());
+                multipartBuilder.addTextBody(paramName, paramValue);
+            }
+            
+            requestEntity = multipartBuilder.build();
+            // Update content type header for multipart
+            req.setHeader(HttpHeaders.CONTENT_TYPE, requestEntity.getContentType());
+        } else {
+            // No statement params - use regular entity template
+            requestEntity = new EntityTemplate(-1, CONTENT_TYPE, contentEncoding, writeCallback);
+        }
+        
+        // Wrap entity with compression if enabled
+        req.setEntity(wrapRequestEntity(requestEntity, lz4Factory, requestConfig));
 
         HttpClientContext context = HttpClientContext.create();
         Number responseTimeout = ClientConfigProperties.SOCKET_OPERATION_TIMEOUT.getOrDefault(requestConfig);
@@ -595,10 +630,7 @@ private void addQueryParams(URIBuilder req, Map<String, Object> requestConfig) {
         if (requestConfig.containsKey(ClientConfigProperties.QUERY_ID.getKey())) {
             req.addParameter(ClickHouseHttpProto.QPARAM_QUERY_ID, requestConfig.get(ClientConfigProperties.QUERY_ID.getKey()).toString());
         }
-        if (requestConfig.containsKey(KEY_STATEMENT_PARAMS)) {
-            Map<?, ?> params = (Map<?, ?>) requestConfig.get(KEY_STATEMENT_PARAMS);
-            params.forEach((k, v) -> req.addParameter("param_" + k, String.valueOf(v)));
-        }
+        // Statement params are now sent as multipart, not query params
 
         boolean clientCompression = ClientConfigProperties.COMPRESS_CLIENT_REQUEST.getOrDefault(requestConfig);
         boolean serverCompression = ClientConfigProperties.COMPRESS_SERVER_RESPONSE.getOrDefault(requestConfig);

client-v2/src/test/java/com/clickhouse/client/HttpTransportTests.java

@@ -43,7 +43,9 @@
 import java.util.Arrays;
 import java.util.Base64;
 import java.util.EnumSet;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Random;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
@@ -1102,6 +1104,55 @@ public void testTimeoutsWithRetry() {
         }
     }
 
+    @Test(groups = {"integration"})
+    public void testStatementParamsSentAsMultipart() throws Exception {
+        if (isCloud()) {
+            return; // mocked server
+        }
+
+        WireMockServer mockServer = new WireMockServer(WireMockConfiguration
+                .options().port(9090).notifier(new ConsoleNotifier(false)));
+        mockServer.start();
+
+        try {
+            // Configure WireMock to capture multipart requests
+            mockServer.addStubMapping(WireMock.post(WireMock.anyUrl())
+                    .withHeader(HttpHeaders.CONTENT_TYPE, WireMock.matching("multipart/form-data.*"))
+                    .withRequestBody(WireMock.containing("param_testParam"))
+                    .withRequestBody(WireMock.containing("testValue"))
+                    .withRequestBody(WireMock.containing("param_anotherParam"))
+                    .withRequestBody(WireMock.containing("123"))
+                    .willReturn(WireMock.aResponse()
+                            .withStatus(HttpStatus.SC_OK)
+                            .withHeader("X-ClickHouse-Summary",
+                                    "{ \"read_bytes\": \"10\", \"read_rows\": \"1\"}")
+                            .withBody("1\n")).build());
+
+            try (Client client = new Client.Builder()
+                    .addEndpoint(Protocol.HTTP, "localhost", mockServer.port(), false)
+                    .setUsername("default")
+                    .setPassword(ClickHouseServerForTest.getPassword())
+                    .compressClientRequest(false)
+                    .build()) {
+
+                // Create query with statement parameters
+                Map<String, Object> queryParams = new HashMap<>();
+                queryParams.put("testParam", "testValue");
+                queryParams.put("anotherParam", 123);
+
+                QueryResponse response = client.query(
+                        "SELECT {testParam:String} as col1, {anotherParam:Int32} as col2",
+                        queryParams).get(10, TimeUnit.SECONDS);
+
+                // Verify the request was made (WireMock will throw if expectations not met)
+                Assert.assertNotNull(response);
+                response.close();
+            }
+        } finally {
+            mockServer.stop();
+        }
+    }
+
     @Test(groups = {"integration"})
     public void testSNIWithCloud() throws Exception {
         if (!isCloud()) {

client-v2/src/test/java/com/clickhouse/client/query/QueryTests.java

@@ -138,6 +138,7 @@ private static void delayForProfiler(long millis) {
         }
     }
 
+    
     @Test(groups = {"integration"})
     public void testSimpleQueryWithTSV() {
         prepareSimpleDataSet();