feat: query parameter for POST variants is serialized into HTTP body

.github/workflows/master.yml

@@ -12,7 +12,7 @@ jobs:
     strategy:
       matrix:
         jdk: [3-openjdk-16-slim, 3-jdk-14, 3-jdk-8-slim]
-        influxdb: [1.1, 1.6, 1.8]
+        influxdb: [1.1, 1.6, 1.8, 2.0]
 
     steps:
       - name: Checkout

.github/workflows/pr.yml

@@ -12,7 +12,7 @@ jobs:
     strategy:
       matrix:
         jdk: [3-openjdk-16-slim, 3-jdk-14, 3-jdk-8-slim]
-        influxdb: [1.1, 1.6, 1.8]
+        influxdb: [1.1, 1.6, 1.8, 2.0]
 
     steps:
       - name: Checkout
@@ -27,3 +27,4 @@ jobs:
 
       - name: codecov
         run: bash <(curl -s https://codecov.io/bash)
+        if: matrix.influxdb != '2.0'

CHANGELOG.md

@@ -2,6 +2,10 @@
 
 ## 2.22 [unreleased]
 
+### Improvements
+
+- Streaming query serializes `'q'` parameter into HTTP body [PR #765](https://github.com/influxdata/influxdb-java/pull/765)
+
 ## 2.21 [2020-12-04]
 
 ### Fixes

compile-and-test.sh

@@ -20,6 +20,12 @@ docker run \
        --publish 8086:8086 \
        --publish 8089:8089/udp \
        --volume ${PWD}/influxdb.conf:/etc/influxdb/influxdb.conf \
+       --env DOCKER_INFLUXDB_INIT_MODE=setup \
+       --env DOCKER_INFLUXDB_INIT_USERNAME=my-user \
+       --env DOCKER_INFLUXDB_INIT_PASSWORD=my-password \
+       --env DOCKER_INFLUXDB_INIT_ORG=my-org \
+       --env DOCKER_INFLUXDB_INIT_BUCKET=my-bucket \
+       --env DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-token \
        influxdb:${INFLUXDB_VERSION}-alpine
 
 echo "Starting Nginx"
@@ -38,6 +44,20 @@ docker run \
 echo "Running tests"
 PROXY_API_URL=http://nginx:8080/influx-api/
 PROXY_UDP_PORT=8080
+if [[ "$INFLUXDB_VERSION" == "2.0" ]]
+then
+    TEST_EXPRESSION="InfluxDB2Test"
+    # Wait to start InfluxDB
+    docker run --link influxdb:influxdb ubuntu:20.04 bash -c "apt-get update \
+      && apt-get install wget --yes \
+      && wget -S --spider --tries=20 --retry-connrefused --waitretry=5 http://influxdb:8086/ping"
+    # Create DBRP Mapping
+    BUCKET_ID=$(docker exec influxdb bash -c "influx bucket list -o my-org -n my-bucket | grep my-bucket | xargs | cut -d ' ' -f 0")
+    docker exec influxdb bash -c "influx v1 dbrp create -o my-org --db mydb --rp autogen --default --bucket-id ${BUCKET_ID}"
+    docker exec influxdb bash -c "influx v1 auth create -o my-org --username my-user --password my-password --read-bucket ${BUCKET_ID} --write-bucket ${BUCKET_ID}"
+else
+    TEST_EXPRESSION="*"
+fi
 
 docker run --rm \
        --volume ${PWD}:/usr/src/mymaven \
@@ -49,7 +69,7 @@ docker run --rm \
        --env INFLUXDB_IP=influxdb \
        --env PROXY_API_URL=${PROXY_API_URL} \
        --env PROXY_UDP_PORT=${PROXY_UDP_PORT} \
-       maven:${MAVEN_JAVA_VERSION} mvn clean install
+       maven:${MAVEN_JAVA_VERSION} mvn clean install -Dtest="${TEST_EXPRESSION}"
 
 docker kill influxdb || true
 docker kill nginx || true

src/main/java/org/influxdb/impl/InfluxDBImpl.java

@@ -642,7 +642,11 @@ public void query(final Query query, final int chunkSize, final BiConsumer<Cance
       call = this.influxDBService.query(getDatabase(query), query.getCommandWithUrlEncoded(), chunkSize,
           boundParameterQuery.getParameterJsonWithUrlEncoded());
     } else {
-      call = this.influxDBService.query(getDatabase(query), query.getCommandWithUrlEncoded(), chunkSize);
+      if (query.requiresPost()) {
+        call = this.influxDBService.query(getDatabase(query), query.getCommandWithUrlEncoded(), chunkSize, null);
+      } else {
+        call = this.influxDBService.query(getDatabase(query), query.getCommandWithUrlEncoded(), chunkSize);
+      }
     }
 
     call.enqueue(new Callback<ResponseBody>() {
@@ -711,15 +715,20 @@ public void onFailure(final Call<ResponseBody> call, final Throwable t) {
    */
   @Override
   public QueryResult query(final Query query, final TimeUnit timeUnit) {
-    Call<QueryResult> call = null;
+    Call<QueryResult> call;
     if (query instanceof BoundParameterQuery) {
         BoundParameterQuery boundParameterQuery = (BoundParameterQuery) query;
         call = this.influxDBService.query(getDatabase(query),
                 TimeUtil.toTimePrecision(timeUnit), query.getCommandWithUrlEncoded(),
                 boundParameterQuery.getParameterJsonWithUrlEncoded());
     } else {
-        call = this.influxDBService.query(getDatabase(query),
-                TimeUtil.toTimePrecision(timeUnit), query.getCommandWithUrlEncoded());
+        if (query.requiresPost()) {
+          call = this.influxDBService.query(getDatabase(query),
+                  TimeUtil.toTimePrecision(timeUnit), query.getCommandWithUrlEncoded(), null);
+        } else {
+          call = this.influxDBService.query(getDatabase(query),
+                  TimeUtil.toTimePrecision(timeUnit), query.getCommandWithUrlEncoded());
+        }
     }
     return executeQuery(call);
   }
@@ -747,7 +756,7 @@ public void deleteDatabase(final String name) {
    */
   @Override
   public List<String> describeDatabases() {
-    QueryResult result = executeQuery(this.influxDBService.query(SHOW_DATABASE_COMMAND_ENCODED));
+    QueryResult result = executeQuery(this.influxDBService.postQuery(SHOW_DATABASE_COMMAND_ENCODED));
     // {"results":[{"series":[{"name":"databases","columns":["name"],"values":[["mydb"]]}]}]}
     // Series [name=databases, columns=[name], values=[[mydb], [unittest_1433605300968]]]
     List<List<Object>> databaseNames = result.getResults().get(0).getSeries().get(0).getValues();

src/main/java/org/influxdb/impl/InfluxDBService.java

@@ -6,6 +6,8 @@
 import okhttp3.ResponseBody;
 import retrofit2.Call;
 import retrofit2.http.Body;
+import retrofit2.http.Field;
+import retrofit2.http.FormUrlEncoded;
 import retrofit2.http.GET;
 import retrofit2.http.POST;
 import retrofit2.http.Query;
@@ -48,27 +50,28 @@ public Call<QueryResult> query(@Query(DB) String db,
       @Query(EPOCH) String epoch, @Query(value = Q, encoded = true) String query);
 
   @POST("query")
+  @FormUrlEncoded
   public Call<QueryResult> query(@Query(DB) String db,
-          @Query(EPOCH) String epoch, @Query(value = Q, encoded = true) String query,
+          @Query(EPOCH) String epoch, @Field(value = Q, encoded = true) String query,
           @Query(value = PARAMS, encoded = true) String params);
 
   @GET("query")
   public Call<QueryResult> query(@Query(DB) String db,
       @Query(value = Q, encoded = true) String query);
 
   @POST("query")
+  @FormUrlEncoded
   public Call<QueryResult> postQuery(@Query(DB) String db,
-      @Query(value = Q, encoded = true) String query);
+      @Field(value = Q, encoded = true) String query);
 
   @POST("query")
+  @FormUrlEncoded
   public Call<QueryResult> postQuery(@Query(DB) String db,
-          @Query(value = Q, encoded = true) String query, @Query(value = PARAMS, encoded = true) String params);
-
-  @GET("query")
-  public Call<QueryResult> query(@Query(value = Q, encoded = true) String query);
+          @Field(value = Q, encoded = true) String query, @Query(value = PARAMS, encoded = true) String params);
 
   @POST("query")
-  public Call<QueryResult> postQuery(@Query(value = Q, encoded = true) String query);
+  @FormUrlEncoded
+  public Call<QueryResult> postQuery(@Field(value = Q, encoded = true) String query);
 
   @Streaming
   @GET("query?chunked=true")
@@ -77,6 +80,7 @@ public Call<ResponseBody> query(@Query(DB) String db, @Query(value = Q, encoded
 
   @Streaming
   @POST("query?chunked=true")
-  public Call<ResponseBody> query(@Query(DB) String db, @Query(value = Q, encoded = true) String query,
+  @FormUrlEncoded
+  public Call<ResponseBody> query(@Query(DB) String db, @Field(value = Q, encoded = true) String query,
           @Query(CHUNK_SIZE) int chunkSize, @Query(value = PARAMS, encoded = true) String params);
 }

src/test/java/org/influxdb/InfluxDB2Test.java

@@ -0,0 +1,57 @@
+package org.influxdb;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.influxdb.dto.Query;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+import org.junit.platform.runner.JUnitPlatform;
+import org.junit.runner.RunWith;
+
+/**
+ * @author Jakub Bednar (30/08/2021 11:31)
+ */
+@RunWith(JUnitPlatform.class)
+@EnabledIfEnvironmentVariable(named = "INFLUXDB_VERSION", matches = "2\\.0")
+public class InfluxDB2Test {
+
+    private InfluxDB influxDB;
+
+    @BeforeEach
+    public void setUp() throws NoSuchFieldException, IllegalAccessException {
+        String url = String.format("http://%s:%s", TestUtils.getInfluxIP(), TestUtils.getInfluxPORT(true));
+        influxDB = InfluxDBFactory
+                .connect(url, "my-user", "my-password")
+                .setDatabase("mydb")
+                .setRetentionPolicy("autogen");
+    }
+
+    @AfterEach
+    public void cleanup() {
+        influxDB.close();
+    }
+
+    @Test
+    public void testQuery() throws InterruptedException {
+
+        String measurement = TestUtils.getRandomMeasurement();
+
+        // prepare data
+        List<String> records = new ArrayList<>();
+        records.add(measurement + ",test=a value=1 1");
+        records.add(measurement + ",test=a value=2 2");
+        influxDB.write(records);
+
+        // query data
+        final CountDownLatch countDownLatch = new CountDownLatch(1);
+        influxDB.query(new Query("SELECT * FROM " + measurement), 2, queryResult -> countDownLatch.countDown());
+
+        Assertions.assertTrue(countDownLatch.await(2, TimeUnit.SECONDS));
+    }
+}

src/test/java/org/influxdb/InfluxDBTest.java

@@ -722,6 +722,16 @@ public void testWriteBatchWithoutPrecision() throws Exception {
 		Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(1).get(0), timeP2);
 		Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(2).get(0), timeP3);
 
+		// WHEN I use the post query
+		queryResult = this.influxDB.query(new Query("SELECT * FROM " + measurement, dbName, true), TimeUnit.NANOSECONDS);
+
+		// THEN result will be same
+		Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().size(), 3);
+		Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().size(), 3);
+		Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(0).get(0), timeP1);
+		Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(1).get(0), timeP2);
+		Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().get(2).get(0), timeP3);
+
 		this.influxDB.query(new Query("DROP DATABASE " + dbName));
 	}
 
@@ -1301,6 +1311,37 @@ public void testChunkingOnFailureConnectionError() throws InterruptedException {
 
   }
 
+	@Test
+    public void testChunkingQueryPost() throws InterruptedException {
+        if (this.influxDB.version().startsWith("0.") || this.influxDB.version().startsWith("1.0")) {
+            // do not test version 0.13 and 1.0
+            return;
+        }
+
+        String dbName = "write_unittest_" + System.currentTimeMillis();
+        this.influxDB.query(new Query("CREATE DATABASE " + dbName));
+        String rp = TestUtils.defaultRetentionPolicy(this.influxDB.version());
+        BatchPoints batchPoints = BatchPoints.database(dbName).retentionPolicy(rp).build();
+        Point point1 = Point.measurement("disk").tag("atag", "a").addField("used", 60L).addField("free", 1L).build();
+        Point point2 = Point.measurement("disk").tag("atag", "b").addField("used", 70L).addField("free", 2L).build();
+        Point point3 = Point.measurement("disk").tag("atag", "c").addField("used", 80L).addField("free", 3L).build();
+        batchPoints.point(point1);
+        batchPoints.point(point2);
+        batchPoints.point(point3);
+        this.influxDB.write(batchPoints);
+
+        CountDownLatch countDownLatch = new CountDownLatch(3);
+
+        Thread.sleep(2000);
+        Query query = new Query("SELECT * FROM disk", dbName, true);
+        this.influxDB.query(query, 2, result -> countDownLatch.countDown());
+
+        boolean await = countDownLatch.await(10, TimeUnit.SECONDS);
+        Assertions.assertTrue(await, "The QueryResults did not arrive!");
+
+        this.influxDB.query(new Query("DROP DATABASE " + dbName));
+    }
+
   @Test
     public void testFlushPendingWritesWhenBatchingEnabled() {
         String dbName = "flush_tests_" + System.currentTimeMillis();

src/test/java/org/influxdb/MessagePackInfluxDBTest.java

@@ -164,6 +164,18 @@ public void testWriteBatchWithoutPrecision() throws Exception {
     value = Double.valueOf(queryResult.getResults().get(0).getSeries().get(0).getValues().get(2).get(0).toString());
     Assertions.assertEquals(value, timeP3);
 
+    // WHEN I use the post query
+    queryResult = this.influxDB.query(new Query("SELECT * FROM " + measurement, dbName, true), TimeUnit.NANOSECONDS);
+
+    // THEN result will be same
+    Assertions.assertEquals(queryResult.getResults().get(0).getSeries().get(0).getValues().size(), 3);
+    value = Double.valueOf(queryResult.getResults().get(0).getSeries().get(0).getValues().get(0).get(0).toString());
+    Assertions.assertEquals(value, timeP1);
+    value = Double.valueOf(queryResult.getResults().get(0).getSeries().get(0).getValues().get(1).get(0).toString());
+    Assertions.assertEquals(value, timeP2);
+    value = Double.valueOf(queryResult.getResults().get(0).getSeries().get(0).getValues().get(2).get(0).toString());
+    Assertions.assertEquals(value, timeP3);
+
     this.influxDB.query(new Query("DROP DATABASE " + dbName));
   }
 

src/test/java/org/influxdb/impl/InfluxDBImplTest.java

@@ -10,9 +10,12 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.platform.runner.JUnitPlatform;
+import org.junit.runner.RunWith;
 
 import okhttp3.OkHttpClient;
 
+@RunWith(JUnitPlatform.class)
 public class InfluxDBImplTest {
 
   private InfluxDB influxDB;

src/test/java/org/influxdb/impl/InfluxDBMapperTest.java

@@ -16,7 +16,10 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.platform.runner.JUnitPlatform;
+import org.junit.runner.RunWith;
 
+@RunWith(JUnitPlatform.class)
 public class InfluxDBMapperTest {
 
   private InfluxDB influxDB;