From 417216bdb0fada800b6ded443271670a66e3ab1e Mon Sep 17 00:00:00 2001 From: Ales Pour Date: Mon, 22 Jan 2024 17:19:02 +0100 Subject: [PATCH 1/8] feat: add SQL query with named parameters --- .../main/java/com/influxdb/v3/IOxExample.java | 15 ++ .../influxdb/v3/client/InfluxDBClient.java | 129 ++++++++++++++++++ .../v3/client/internal/FlightSqlClient.java | 9 +- .../client/internal/InfluxDBClientImpl.java | 76 ++++++++++- .../com/influxdb/v3/client/ITQueryWrite.java | 29 +++- 5 files changed, 247 insertions(+), 11 deletions(-) diff --git a/examples/src/main/java/com/influxdb/v3/IOxExample.java b/examples/src/main/java/com/influxdb/v3/IOxExample.java index 1799301..23d4568 100644 --- a/examples/src/main/java/com/influxdb/v3/IOxExample.java +++ b/examples/src/main/java/com/influxdb/v3/IOxExample.java @@ -22,6 +22,7 @@ package com.influxdb.v3; import java.time.Instant; +import java.util.Map; import java.util.stream.Stream; import com.influxdb.v3.client.InfluxDBClient; @@ -73,6 +74,20 @@ public static void main(final String[] args) throws Exception { System.out.printf("--------------------------------------------------------%n%n"); + // + // Query by parametrized SQL + // + System.out.printf("--------------------------------------------------------%n"); + System.out.printf("| %-8s | %-8s | %-30s |%n", "location", "value", "time"); + System.out.printf("--------------------------------------------------------%n"); + + String sqlWithParameters = "select time,location,value from temperature where location=$location order by time desc limit 10"; + try (Stream stream = client.query(sqlWithParameters, Map.of("location", "north"))) { + stream.forEach(row -> System.out.printf("| %-8s | %-8s | %-30s |%n", row[1], row[2], row[0])); + } + + System.out.printf("--------------------------------------------------------%n%n"); + // // Query by InfluxQL // diff --git a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java index 2eecd5d..a02ed54 100644 --- a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java +++ b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java @@ -120,6 +120,26 @@ public interface InfluxDBClient extends AutoCloseable { @Nonnull Stream query(@Nonnull final String query); + /** + * Query data from InfluxDB IOx using FlightSQL. + *

+ * The result stream should be closed after use, you can use try-resource pattern to close it automatically: + * 

+     * try (Stream<Object[]> rows = client.query("select * from cpu where host=$host",
+     *                                                 Map.of("host", "server-a")) {
+     *      rows.forEach(row -> {
+     *          // process row
+     *      }
+     * });
+     *
+ * + * @param query the SQL query string to execute, cannot be null + * @param parameters query named parameters + * @return Batches of rows returned by the query + */ + @Nonnull + Stream query(@Nonnull final String query, @Nonnull final Map parameters); + /** * Query data from InfluxDB IOx using FlightSQL. *

@@ -139,6 +159,29 @@ public interface InfluxDBClient extends AutoCloseable { @Nonnull Stream query(@Nonnull final String query, @Nonnull final QueryOptions options); + /** + * Query data from InfluxDB IOx using FlightSQL. + *

+ * The result stream should be closed after use, you can use try-resource pattern to close it automatically: + * 

+     * try (Stream<Object[]> rows = client.query("select * from cpu where host=$host",
+     *                                                 Map.of("host", "server-a"), options)) {
+     *      rows.forEach(row -> {
+     *          // process row
+     *      }
+     * });
+     *
+ * + * @param query the SQL query string to execute, cannot be null + * @param parameters query named parameters + * @param options the options for querying data from InfluxDB + * @return Batches of rows returned by the query + */ + @Nonnull + Stream query(@Nonnull final String query, + @Nonnull final Map parameters, + @Nonnull final QueryOptions options); + /** * Query data from InfluxDB IOx into Point structure using FlightSQL. *

@@ -157,6 +200,26 @@ public interface InfluxDBClient extends AutoCloseable { @Nonnull Stream queryPoints(@Nonnull final String query); + /** + * Query data from InfluxDB IOx into Point structure using FlightSQL. + *

+ * The result stream should be closed after use, you can use try-resource pattern to close it automatically: + * 

+     * try (Stream<PointValues> rows = client.queryPoints("select * from cpu where host=$host",
+     *                                                          Map.of("host", "server-a"))) {
+     *      rows.forEach(row -> {
+     *          // process row
+     *      }
+     * });
+     *
+ * + * @param query the SQL query string to execute, cannot be null + * @param parameters query named parameters + * @return Batches of PointValues returned by the query + */ + @Nonnull + Stream queryPoints(@Nonnull final String query, @Nonnull final Map parameters); + /** * Query data from InfluxDB IOx into Point structure using FlightSQL. *

@@ -176,6 +239,30 @@ public interface InfluxDBClient extends AutoCloseable { @Nonnull Stream queryPoints(@Nonnull final String query, @Nonnull final QueryOptions options); + /** + * Query data from InfluxDB IOx into Point structure using FlightSQL. + *

+ * The result stream should be closed after use, you can use try-resource pattern to close it automatically: + * 

+     * try (Stream<PointValues> rows = client.queryPoints("select * from cpu where host=$host",
+     *                                                          Map.of("host", "server-a"),
+     *                                                          options)) {
+     *      rows.forEach(row -> {
+     *          // process row
+     *      }
+     * });
+     *
+ * + * @param query the SQL query string to execute, cannot be null + * @param parameters query named parameters + * @param options the options for querying data from InfluxDB + * @return Batches of PointValues returned by the query + */ + @Nonnull + Stream queryPoints(@Nonnull final String query, + @Nonnull final Map parameters, + @Nonnull final QueryOptions options); + /** * Query data from InfluxDB IOx using FlightSQL. *

@@ -194,6 +281,26 @@ public interface InfluxDBClient extends AutoCloseable { @Nonnull Stream queryBatches(@Nonnull final String query); + /** + * Query data from InfluxDB IOx using FlightSQL. + *

+ * The result stream should be closed after use, you can use try-resource pattern to close it automatically: + * 

+     * try (Stream<VectorSchemaRoot> batches = client.queryBatches("select * from cpu where host=$host",
+     *                                                                   Map.of("host", "server-a"))) {
+     *      batches.forEach(batch -> {
+     *          // process batch
+     *      }
+     * });
+     *
+ * + * @param query the SQL query string to execute, cannot be null + * @param parameters query named parameters + * @return Batches of rows returned by the query + */ + @Nonnull + Stream queryBatches(@Nonnull final String query, @Nonnull final Map parameters); + /** * Query data from InfluxDB IOx using FlightSQL. * 
@@ -211,6 +318,28 @@ public interface InfluxDBClient extends AutoCloseable {
     @Nonnull
     Stream queryBatches(@Nonnull final String query, @Nonnull final QueryOptions options);
 
+    /**
+     * Query data from InfluxDB IOx using FlightSQL.
+     * 
+     * try (Stream<VectorSchemaRoot> batches = client.queryBatches("select * from cpu where host=$host",
+     *                                                                   Map.of("host", "server-a"),
+     *                                                                   options)) {
+     *      batches.forEach(batch -> {
+     *          // process batch
+     *      }
+     * });
+     * 

+     *
+     * @param query      the SQL query string to execute, cannot be null
+     * @param parameters query named parameters
+     * @param options the options for querying data from InfluxDB
+     * @return Batches of rows returned by the query
+     */
+    @Nonnull
+    Stream queryBatches(@Nonnull final String query,
+                                          @Nonnull final Map parameters,
+                                          @Nonnull final QueryOptions options);
+
     /**
      * Creates a new instance of the {@link InfluxDBClient} for interacting with an InfluxDB server, simplifying
      * common operations such as writing, querying.
diff --git a/src/main/java/com/influxdb/v3/client/internal/FlightSqlClient.java b/src/main/java/com/influxdb/v3/client/internal/FlightSqlClient.java
index 8ff8b8c..c37add1 100644
--- a/src/main/java/com/influxdb/v3/client/internal/FlightSqlClient.java
+++ b/src/main/java/com/influxdb/v3/client/internal/FlightSqlClient.java
@@ -91,14 +91,19 @@ final class FlightSqlClient implements AutoCloseable {
     @Nonnull
     Stream execute(@Nonnull final String query,
                                      @Nonnull final String database,
-                                     @Nonnull final QueryType queryType) {
+                                     @Nonnull final QueryType queryType,
+                                     @Nonnull final Map queryParameters) {
 
-        Map ticketData = new HashMap<>() {{
+        Map ticketData = new HashMap<>() {{
             put("database", database);
             put("sql_query", query);
             put("query_type", queryType.name().toLowerCase());
         }};
 
+        if (queryParameters.size() > 0) {
+            ticketData.put("params", queryParameters);
+        }
+
         String json;
         try {
             json = objectMapper.writeValueAsString(ticketData);
diff --git a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
index d14d626..ab8aba8 100644
--- a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
+++ b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
@@ -29,6 +29,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -62,6 +63,16 @@ public final class InfluxDBClientImpl implements InfluxDBClient {
     private static final String DATABASE_REQUIRED_MESSAGE = "Please specify the 'Database' as a method parameter "
             + "or use default configuration at 'ClientConfig.database'.";
 
+    private static final Map NO_PARAMETERS = Map.of();
+    private static final List ALLOWED_NAMED_PARAMETER_TYPES = List.of(
+            String.class,
+            Integer.class,
+            Long.class,
+            Float.class,
+            Double.class,
+            Boolean.class
+    );
+
     private boolean closed = false;
     private final ClientConfig config;
 
@@ -136,13 +147,27 @@ public void writePoints(@Nonnull final List points, @Nonnull final WriteO
     @Nonnull
     @Override
     public Stream query(@Nonnull final String query) {
-        return query(query, QueryOptions.DEFAULTS);
+        return query(query, NO_PARAMETERS, QueryOptions.DEFAULTS);
     }
 
     @Nonnull
     @Override
     public Stream query(@Nonnull final String query, @Nonnull final QueryOptions options) {
-        return queryData(query, options)
+        return query(query, NO_PARAMETERS, options);
+    }
+
+    @Nonnull
+    @Override
+    public Stream query(@Nonnull final String query, @Nonnull final Map parameters) {
+        return query(query, parameters, QueryOptions.DEFAULTS);
+    }
+
+    @Nonnull
+    @Override
+    public Stream query(@Nonnull final String query,
+                                  @Nonnull final Map parameters,
+                                  @Nonnull final QueryOptions options) {
+        return queryData(query, parameters, options)
                 .flatMap(vector -> {
                     List fieldVectors = vector.getFieldVectors();
                     return IntStream
@@ -161,13 +186,27 @@ public Stream query(@Nonnull final String query, @Nonnull final QueryO
     @Nonnull
     @Override
     public Stream queryPoints(@Nonnull final String query) {
-        return queryPoints(query, QueryOptions.DEFAULTS);
+        return queryPoints(query, NO_PARAMETERS, QueryOptions.DEFAULTS);
     }
 
     @Nonnull
     @Override
     public Stream queryPoints(@Nonnull final String query, @Nonnull final QueryOptions options) {
-        return queryData(query, options)
+        return queryPoints(query, NO_PARAMETERS, options);
+    }
+
+    @Nonnull
+    @Override
+    public Stream queryPoints(@Nonnull final String query, @Nonnull final Map parameters) {
+        return queryPoints(query, parameters, QueryOptions.DEFAULTS);
+    }
+
+    @Nonnull
+    @Override
+    public Stream queryPoints(@Nonnull final String query,
+                                           @Nonnull final Map parameters,
+                                           @Nonnull final QueryOptions options) {
+        return queryData(query, parameters, options)
                 .flatMap(vector -> {
                     List fieldVectors = vector.getFieldVectors();
                     return IntStream
@@ -180,14 +219,28 @@ public Stream queryPoints(@Nonnull final String query, @Nonnull fin
     @Nonnull
     @Override
     public Stream queryBatches(@Nonnull final String query) {
-        return queryBatches(query, QueryOptions.DEFAULTS);
+        return queryBatches(query, NO_PARAMETERS, QueryOptions.DEFAULTS);
+    }
+
+    @Nonnull
+    @Override
+    public Stream queryBatches(@Nonnull final String query, @Nonnull final QueryOptions options) {
+        return queryBatches(query, NO_PARAMETERS, options);
     }
 
     @Nonnull
     @Override
     public Stream queryBatches(@Nonnull final String query,
+                                                 @Nonnull final Map parameters) {
+        return queryBatches(query, parameters, QueryOptions.DEFAULTS);
+    }
+
+    @Nonnull
+    @Override
+    public Stream queryBatches(@Nonnull final String query,
+                                                 @Nonnull final Map parameters,
                                                  @Nonnull final QueryOptions options) {
-        return queryData(query, options);
+        return queryData(query, parameters, options);
     }
 
     @Override
@@ -258,9 +311,11 @@ private  void writeData(@Nonnull final List data, @Nonnull final WriteOpti
 
     @Nonnull
     private Stream queryData(@Nonnull final String query,
+                                               @Nonnull final Map parameters,
                                                @Nonnull final QueryOptions options) {
 
         Arguments.checkNonEmpty(query, "query");
+        Arguments.checkNotNull(parameters, "parameters");
         Arguments.checkNotNull(options, "options");
 
         if (closed) {
@@ -272,7 +327,14 @@ private Stream queryData(@Nonnull final String query,
             throw new IllegalStateException(DATABASE_REQUIRED_MESSAGE);
         }
 
-        return flightSqlClient.execute(query, database, options.queryTypeSafe());
+        parameters.forEach((k, v) -> {
+            if (!Objects.isNull(v) && !ALLOWED_NAMED_PARAMETER_TYPES.contains(v.getClass())) {
+                throw new IllegalArgumentException(String.format("The parameter %s value has unsupported type: %s",
+                        k, v.getClass()));
+            }
+        });
+
+        return flightSqlClient.execute(query, database, options.queryTypeSafe(), parameters);
     }
 
     @Nonnull
diff --git a/src/test/java/com/influxdb/v3/client/ITQueryWrite.java b/src/test/java/com/influxdb/v3/client/ITQueryWrite.java
index 638ac85..917c048 100644
--- a/src/test/java/com/influxdb/v3/client/ITQueryWrite.java
+++ b/src/test/java/com/influxdb/v3/client/ITQueryWrite.java
@@ -25,6 +25,7 @@
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -59,7 +60,7 @@ void queryWrite() {
         client = getInstance();
 
         String measurement = "integration_test";
-        int testId = (int) System.currentTimeMillis();
+        long testId = System.currentTimeMillis();
         client.writeRecord(measurement + ",type=used value=123.0,testId=" + testId);
 
         String sql = String.format("SELECT value FROM %s WHERE \"testId\"=%d", measurement, testId);
@@ -124,7 +125,7 @@ void queryWriteGzip() {
                 .build());
 
         String measurement = "integration_test";
-        int testId = (int) System.currentTimeMillis();
+        long testId = System.currentTimeMillis();
         client.writeRecord(measurement + ",type=used value=123.0,testId=" + testId);
 
         String sql = String.format("SELECT value FROM %s WHERE \"testId\"=%d", measurement, testId);
@@ -136,6 +137,30 @@ void queryWriteGzip() {
         }
     }
 
+    @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_URL", matches = ".*")
+    @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_TOKEN", matches = ".*")
+    @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_DATABASE", matches = ".*")
+    @Test
+    void queryWriteParameters() {
+        client = InfluxDBClient.getInstance(new ClientConfig.Builder()
+                .host(System.getenv("TESTING_INFLUXDB_URL"))
+                .token(System.getenv("TESTING_INFLUXDB_TOKEN").toCharArray())
+                .database(System.getenv("TESTING_INFLUXDB_DATABASE"))
+                .build());
+
+        String measurement = "integration_test";
+        long testId = System.currentTimeMillis();
+        client.writeRecord(measurement + ",type=used value=124.0,testId=" + testId);
+
+        String sql = String.format("SELECT value FROM %s WHERE \"testId\"=$id", measurement);
+        try (Stream stream = client.query(sql, Map.of("id", testId))) {
+            stream.forEach(row -> {
+                Assertions.assertThat(row).hasSize(1);
+                Assertions.assertThat(row[0]).isEqualTo(124.0);
+            });
+        }
+    }
+
     @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_URL", matches = ".*")
     @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_TOKEN", matches = ".*")
     @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_DATABASE", matches = ".*")

From 59711fe75fad87a71cec30fdd48f39ab420ad508 Mon Sep 17 00:00:00 2001
From: Ales Pour 
Date: Mon, 22 Jan 2024 17:30:42 +0100
Subject: [PATCH 2/8] style: fix example formatting

---
 examples/src/main/java/com/influxdb/v3/IOxExample.java | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/examples/src/main/java/com/influxdb/v3/IOxExample.java b/examples/src/main/java/com/influxdb/v3/IOxExample.java
index 23d4568..ee77241 100644
--- a/examples/src/main/java/com/influxdb/v3/IOxExample.java
+++ b/examples/src/main/java/com/influxdb/v3/IOxExample.java
@@ -81,7 +81,8 @@ public static void main(final String[] args) throws Exception {
             System.out.printf("| %-8s | %-8s | %-30s |%n", "location", "value", "time");
             System.out.printf("--------------------------------------------------------%n");
 
-            String sqlWithParameters = "select time,location,value from temperature where location=$location order by time desc limit 10";
+            String sqlWithParameters =
+                    "select time,location,value from temperature where location=$location order by time desc limit 10";
             try (Stream stream = client.query(sqlWithParameters, Map.of("location", "north"))) {
                 stream.forEach(row -> System.out.printf("| %-8s | %-8s | %-30s |%n", row[1], row[2], row[0]));
             }

From 754641342a1fa34bfa885ed475fec4045375a182 Mon Sep 17 00:00:00 2001
From: Jakub Bednar 
Date: Fri, 23 Feb 2024 09:48:28 +0100
Subject: [PATCH 3/8] docs: update CHANGELOG.md

---
 CHANGELOG.md                                           | 5 +++++
 examples/src/main/java/com/influxdb/v3/IOxExample.java | 2 +-
 2 files changed, 6 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 58e35d6..47834f1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,10 @@
 ## 0.6.0 [unreleased]
 
+### Features
+
+1. [#94](https://github.com/InfluxCommunity/influxdb3-java/pull/94): Add support for named query parameters
+
+
 ## 0.5.1 [2024-02-01]
 
 Resync artifacts with Maven Central.
diff --git a/examples/src/main/java/com/influxdb/v3/IOxExample.java b/examples/src/main/java/com/influxdb/v3/IOxExample.java
index ee77241..2a3b2ad 100644
--- a/examples/src/main/java/com/influxdb/v3/IOxExample.java
+++ b/examples/src/main/java/com/influxdb/v3/IOxExample.java
@@ -77,7 +77,7 @@ public static void main(final String[] args) throws Exception {
             //
             // Query by parametrized SQL
             //
-            System.out.printf("--------------------------------------------------------%n");
+            System.out.printf("--------------------Parametrized SQL--------------------%n");
             System.out.printf("| %-8s | %-8s | %-30s |%n", "location", "value", "time");
             System.out.printf("--------------------------------------------------------%n");
 

From e0b29a128286f5b89c426652c412dc3729350e6c Mon Sep 17 00:00:00 2001
From: Jakub Bednar 
Date: Fri, 23 Feb 2024 09:52:47 +0100
Subject: [PATCH 4/8] docs: update README.md

---
 CHANGELOG.md |  1 -
 README.md    | 14 ++++++++++++++
 2 files changed, 14 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 47834f1..61a33d7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,7 +4,6 @@
 
 1. [#94](https://github.com/InfluxCommunity/influxdb3-java/pull/94): Add support for named query parameters
 
-
 ## 0.5.1 [2024-02-01]
 
 Resync artifacts with Maven Central.
diff --git a/README.md b/README.md
index 2107237..2af9931 100644
--- a/README.md
+++ b/README.md
@@ -120,6 +120,20 @@ try (Stream stream = client.query(sql)) {
 
 System.out.printf("--------------------------------------------------------%n%n");
 
+//
+// Query by parametrized SQL
+//
+System.out.printf("--------------------Parametrized SQL--------------------%n");
+System.out.printf("| %-8s | %-8s | %-30s |%n", "location", "value", "time");
+System.out.printf("--------------------------------------------------------%n");
+
+String sqlParams = "select time,location,value from temperature where location=$location order by time desc limit 10";
+try (Stream stream = client.query(sqlParams, Map.of("location", "north"))) {
+    stream.forEach(row -> System.out.printf("| %-8s | %-8s | %-30s |%n", row[1], row[2], row[0]));
+}
+
+System.out.printf("--------------------------------------------------------%n%n");
+
 //
 // Query by InfluxQL
 //

From f2e21c4d2531e2bc521d6817f39603c0f2b4b9d2 Mon Sep 17 00:00:00 2001
From: Jakub Bednar 
Date: Fri, 23 Feb 2024 10:21:24 +0100
Subject: [PATCH 5/8] chore: improve test coverage

---
 .../com/influxdb/v3/client/internal/InfluxDBClientImpl.java     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
index ab8aba8..ae4bcbe 100644
--- a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
+++ b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
@@ -186,7 +186,7 @@ public Stream query(@Nonnull final String query,
     @Nonnull
     @Override
     public Stream queryPoints(@Nonnull final String query) {
-        return queryPoints(query, NO_PARAMETERS, QueryOptions.DEFAULTS);
+        return queryPoints(query, QueryOptions.DEFAULTS);
     }
 
     @Nonnull

From 10db81e3054e637f99abd1840dce6bbde8d519aa Mon Sep 17 00:00:00 2001
From: Jakub Bednar 
Date: Fri, 23 Feb 2024 10:27:42 +0100
Subject: [PATCH 6/8] chore: improve test coverage

---
 .../com/influxdb/v3/client/internal/InfluxDBClientImpl.java     | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
index ae4bcbe..a3ed6af 100644
--- a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
+++ b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
@@ -219,7 +219,7 @@ public Stream queryPoints(@Nonnull final String query,
     @Nonnull
     @Override
     public Stream queryBatches(@Nonnull final String query) {
-        return queryBatches(query, NO_PARAMETERS, QueryOptions.DEFAULTS);
+        return queryBatches(query, QueryOptions.DEFAULTS);
     }
 
     @Nonnull

From 594de152c7738b68265644e17285b0335466a2c2 Mon Sep 17 00:00:00 2001
From: Jakub Bednar 
Date: Fri, 23 Feb 2024 10:47:08 +0100
Subject: [PATCH 7/8] chore: improve test coverage

---
 .../v3/client/internal/InfluxDBClientImpl.java    |  2 +-
 .../influxdb/v3/client/InfluxDBClientTest.java    | 15 +++++++++++++++
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
index a3ed6af..b64d5e3 100644
--- a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
+++ b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java
@@ -64,7 +64,7 @@ public final class InfluxDBClientImpl implements InfluxDBClient {
             + "or use default configuration at 'ClientConfig.database'.";
 
     private static final Map NO_PARAMETERS = Map.of();
-    private static final List ALLOWED_NAMED_PARAMETER_TYPES = List.of(
+    private static final List> ALLOWED_NAMED_PARAMETER_TYPES = List.of(
             String.class,
             Integer.class,
             Long.class,
diff --git a/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java b/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java
index c5ef703..2c9fdde 100644
--- a/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java
+++ b/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java
@@ -101,4 +101,19 @@ void withDefaultTags() throws Exception {
             Assertions.assertThat(client).isNotNull();
         }
     }
+
+    @Test
+    public void unsupportedQueryParams() throws Exception {
+        try (InfluxDBClient client = InfluxDBClient.getInstance("http://localhost:8086",
+                "my-token".toCharArray(), "my-database")) {
+
+            String query = "select * from cpu where client=$client";
+            Map parameters = Map.of("client", client);
+
+            Assertions.assertThatThrownBy(() -> client.queryPoints(query, parameters))
+                    .isInstanceOf(IllegalArgumentException.class)
+                    .hasMessage("The parameter client value has unsupported type: "
+                            + "class com.influxdb.v3.client.internal.InfluxDBClientImpl");
+        }
+    }
 }

From 3601a5e2f1bb65a4bb45e9ca6f483a796e3bfa29 Mon Sep 17 00:00:00 2001
From: Jakub Bednar 
Date: Fri, 23 Feb 2024 13:42:11 +0100
Subject: [PATCH 8/8] chore: improve test coverage

---
 src/test/java/com/influxdb/v3/client/ITQueryWrite.java | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/src/test/java/com/influxdb/v3/client/ITQueryWrite.java b/src/test/java/com/influxdb/v3/client/ITQueryWrite.java
index 917c048..ca6462d 100644
--- a/src/test/java/com/influxdb/v3/client/ITQueryWrite.java
+++ b/src/test/java/com/influxdb/v3/client/ITQueryWrite.java
@@ -152,13 +152,21 @@ void queryWriteParameters() {
         long testId = System.currentTimeMillis();
         client.writeRecord(measurement + ",type=used value=124.0,testId=" + testId);
 
+        Map parameters = Map.of("id", testId);
         String sql = String.format("SELECT value FROM %s WHERE \"testId\"=$id", measurement);
-        try (Stream stream = client.query(sql, Map.of("id", testId))) {
+        try (Stream stream = client.query(sql, parameters)) {
             stream.forEach(row -> {
                 Assertions.assertThat(row).hasSize(1);
                 Assertions.assertThat(row[0]).isEqualTo(124.0);
             });
         }
+        try (Stream stream = client.queryPoints(sql, parameters)) {
+            stream.forEach(row -> Assertions.assertThat(row.getField("value")).isEqualTo(124.0));
+        }
+        try (Stream batches = client.queryBatches(sql, parameters)) {
+
+            Assertions.assertThat(batches.count()).isGreaterThanOrEqualTo(1);
+        }
     }
 
     @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_URL", matches = ".*")