From bf311f1251b7be968eca3f406cb91a541061606c Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Fri, 20 Dec 2024 11:04:31 +0700 Subject: [PATCH 01/13] feat: add queryRows function --- .../influxdb/v3/client/InfluxDBClient.java | 82 +++++++++++++++++++ .../client/internal/InfluxDBClientImpl.java | 33 ++++++++ .../internal/VectorSchemaRootConverter.java | 23 ++++++ .../v3/client/InfluxDBClientTest.java | 59 ++++++++++++- 4 files changed, 195 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java index 2a11efbd..c6cd89b8 100644 --- a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java +++ b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java @@ -182,6 +182,88 @@ Stream query(@Nonnull final String query, @Nonnull final Map parameters, @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<Map<String, Object>> rows = client.query("select * from cpu where host=$host",;
+     *                                                 Map.of("host", "server-a"), options)) {
+     *      rows.forEach(row -> {
+     *          // process row
+     *      }
+     * });
+     *
+ * + * @param query the query string to execute, cannot be null + * @return Batches of rows returned by the query + */ + @Nonnull + Stream> queryRows(@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<Map<String, Object>> rows = client.query("select * from cpu where host=$host",;
+     *                                                 Map.of("host", "server-a"), options)) {
+     *      rows.forEach(row -> {
+     *          // process row
+     *      }
+     * });
+     *
+ * + * @param query the query string to execute, cannot be null + * @param parameters query named parameters + * @return Batches of rows returned by the query + */ + @Nonnull + Stream> queryRows(@Nonnull final String query, @Nonnull final Map parameters); + + /** + * 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<Map<String, Object>> rows = client.query("select * from cpu where host=$host",;
+     *                                                 Map.of("host", "server-a"), options)) {
+     *      rows.forEach(row -> {
+     *          // process row
+     *      }
+     * });
+     *
+ * + * @param query the query string to execute, cannot be null + * @param options the options for querying data from InfluxDB + * @return Batches of rows returned by the query + */ + @Nonnull + Stream> queryRows(@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<Map<String, Object>> rows = client.query("select * from cpu where host=$host",;
+     *                                                 Map.of("host", "server-a"), options)) {
+     *      rows.forEach(row -> {
+     *          // process row
+     *      }
+     * });
+     *
+ * + * @param query the 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> queryRows(@Nonnull final String query, + @Nonnull final Map parameters, + @Nonnull final QueryOptions options); + /** * Query data from InfluxDB IOx into Point structure using FlightSQL. *

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 6d67b7d0..c060756a 100644 --- a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java +++ b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java @@ -189,6 +189,39 @@ public Stream query(@Nonnull final String query, ))); } + @Nonnull + @Override + public Stream> queryRows(@Nonnull String query) { + return queryRows(query, NO_PARAMETERS, QueryOptions.DEFAULTS); + } + + @Nonnull + @Override + public Stream> queryRows(@Nonnull String query, @Nonnull Map parameters) { + return queryRows(query, parameters, QueryOptions.DEFAULTS); + } + + @Nonnull + @Override + public Stream> queryRows(@Nonnull String query, @Nonnull QueryOptions options) { + return queryRows(query, NO_PARAMETERS, options); + } + + @Nonnull + @Override + public Stream> queryRows(@Nonnull final String query, + @Nonnull final Map parameters, + @Nonnull final QueryOptions options) { + return queryData(query, parameters, options) + .flatMap(vector -> IntStream.range(0, vector.getRowCount()) + .mapToObj(rowNumber -> + VectorSchemaRootConverter.INSTANCE + .getMapFromVectorSchemaRoot( + vector, + rowNumber + ))); + } + @Nonnull @Override public Stream queryPoints(@Nonnull final String query) { diff --git a/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java b/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java index b2ce9587..bb611924 100644 --- a/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java +++ b/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java @@ -23,7 +23,9 @@ import java.math.BigInteger; import java.time.LocalDateTime; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.logging.Logger; import javax.annotation.Nonnull; @@ -195,4 +197,25 @@ public Object[] getArrayObjectFromVectorSchemaRoot(@Nonnull final VectorSchemaRo return row; } + + /** + * Get a Map from VectorSchemaRoot. + * + * @param vector The data return from InfluxDB. + * @param rowNumber The row number of data + * @return A Map represents a row of data + */ + public Map getMapFromVectorSchemaRoot(VectorSchemaRoot vector, int rowNumber) { + Map row = new HashMap<>(); + for (FieldVector fieldVector : vector.getFieldVectors()) { + Object mappedValue = getMappedValue( + fieldVector.getField(), + fieldVector.getObject(rowNumber) + ); + row.put(fieldVector.getName(), mappedValue); + + } + + return row; + } } diff --git a/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java b/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java index f1e868a2..d4d10cfe 100644 --- a/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java +++ b/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java @@ -138,7 +138,7 @@ public void testQuery() throws Exception { String uuid = UUID.randomUUID().toString(); long timestamp = Instant.now().getEpochSecond(); String record = String.format( - "host10,tag=empty " + "host12,tag=empty " + "name=\"intel\"," + "mem_total=2048," + "disk_free=100i," @@ -151,7 +151,7 @@ public void testQuery() throws Exception { client.writeRecord(record, new WriteOptions(null, WritePrecision.S, null)); Map parameters = Map.of("testId", uuid); - String sql = "Select * from host10 where \"testId\"=$testId"; + String sql = "Select * from host12 where \"testId\"=$testId"; try (Stream stream = client.query(sql, parameters)) { stream.findFirst() .ifPresent(objects -> { @@ -167,10 +167,65 @@ public void testQuery() throws Exception { Assertions.assertThat(objects[3].getClass()).isEqualTo(String.class); Assertions.assertThat(objects[3]).isEqualTo("intel"); + Assertions.assertThat(objects[4].getClass()).isEqualTo(String.class); + Assertions.assertThat(objects[4]).isEqualTo("empty"); + Assertions.assertThat(objects[7].getClass()).isEqualTo(BigInteger.class); Assertions.assertThat(objects[7]).isEqualTo(BigInteger.valueOf(timestamp * 1_000_000_000)); }); } } } + + @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_URL", matches = ".*") + @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_TOKEN", matches = ".*") + @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_DATABASE", matches = ".*") + @Test + public void testQueryRows() throws Exception { + try (InfluxDBClient client = InfluxDBClient.getInstance( + System.getenv("TESTING_INFLUXDB_URL"), + System.getenv("TESTING_INFLUXDB_TOKEN").toCharArray(), + System.getenv("TESTING_INFLUXDB_DATABASE"), + null)) { + String uuid = UUID.randomUUID().toString(); + long timestamp = Instant.now().getEpochSecond(); + String record = String.format( + "host12,tag=tagValue " + + "name=\"intel\"," + + "mem_total=2048," + + "disk_free=100i," + + "temperature=100.86," + + "isActive=true," + + "testId=\"%s\" %d", + uuid, + timestamp + ); + client.writeRecord(record, new WriteOptions(null, WritePrecision.S, null)); + + Map parameters = Map.of("testId", uuid); + String sql = "Select * from host12 where \"testId\"=$testId"; + try (Stream> stream = client.queryRows(sql, parameters)) { + stream.findFirst() + .ifPresent(map -> { + Assertions.assertThat(map.get("tag").getClass()).isEqualTo(String.class); + Assertions.assertThat(map.get("tag")).isEqualTo("tagValue"); + + Assertions.assertThat(map.get("name").getClass()).isEqualTo(String.class); + Assertions.assertThat(map.get("name")).isEqualTo("intel"); + + Assertions.assertThat(map.get("mem_total").getClass()).isEqualTo(Double.class); + Assertions.assertThat(map.get("mem_total")).isEqualTo(2048.0); + + Assertions.assertThat(map.get("disk_free").getClass()).isEqualTo(Long.class); + Assertions.assertThat(map.get("disk_free")).isEqualTo(100L); + + Assertions.assertThat(map.get("isActive").getClass()).isEqualTo(Boolean.class); + Assertions.assertThat(map.get("isActive")).isEqualTo(true); + + Assertions.assertThat(map.get("time").getClass()).isEqualTo(BigInteger.class); + Assertions.assertThat(map.get("time")).isEqualTo(BigInteger.valueOf(timestamp * 1_000_000_000)); + }); + } + } + } } From 3a1132bbf2f4c6c1bcd64b91011dfcaae946b571 Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Fri, 20 Dec 2024 11:12:30 +0700 Subject: [PATCH 02/13] fix: linter --- .../com/influxdb/v3/client/internal/InfluxDBClientImpl.java | 6 +++--- .../v3/client/internal/VectorSchemaRootConverter.java | 2 +- .../java/com/influxdb/v3/client/InfluxDBClientTest.java | 3 ++- 3 files changed, 6 insertions(+), 5 deletions(-) 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 c060756a..4ed9a571 100644 --- a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java +++ b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java @@ -191,19 +191,19 @@ public Stream query(@Nonnull final String query, @Nonnull @Override - public Stream> queryRows(@Nonnull String query) { + public Stream> queryRows(@Nonnull final String query) { return queryRows(query, NO_PARAMETERS, QueryOptions.DEFAULTS); } @Nonnull @Override - public Stream> queryRows(@Nonnull String query, @Nonnull Map parameters) { + public Stream> queryRows(@Nonnull final String query, @Nonnull final Map parameters) { return queryRows(query, parameters, QueryOptions.DEFAULTS); } @Nonnull @Override - public Stream> queryRows(@Nonnull String query, @Nonnull QueryOptions options) { + public Stream> queryRows(@Nonnull final String query, @Nonnull final QueryOptions options) { return queryRows(query, NO_PARAMETERS, options); } diff --git a/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java b/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java index bb611924..4c2bb8b3 100644 --- a/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java +++ b/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java @@ -205,7 +205,7 @@ public Object[] getArrayObjectFromVectorSchemaRoot(@Nonnull final VectorSchemaRo * @param rowNumber The row number of data * @return A Map represents a row of data */ - public Map getMapFromVectorSchemaRoot(VectorSchemaRoot vector, int rowNumber) { + public Map getMapFromVectorSchemaRoot(@Nonnull final VectorSchemaRoot vector, final int rowNumber) { Map row = new HashMap<>(); for (FieldVector fieldVector : vector.getFieldVectors()) { Object mappedValue = getMappedValue( diff --git a/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java b/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java index d4d10cfe..3294dcc3 100644 --- a/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java +++ b/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java @@ -223,7 +223,8 @@ public void testQueryRows() throws Exception { Assertions.assertThat(map.get("isActive")).isEqualTo(true); Assertions.assertThat(map.get("time").getClass()).isEqualTo(BigInteger.class); - Assertions.assertThat(map.get("time")).isEqualTo(BigInteger.valueOf(timestamp * 1_000_000_000)); + Assertions.assertThat(map.get("time")) + .isEqualTo(BigInteger.valueOf(timestamp * 1_000_000_000)); }); } } From e6a86b75a3c66f6fd4b617307a9ec6d7b800f1a9 Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Fri, 20 Dec 2024 11:17:43 +0700 Subject: [PATCH 03/13] fix: linter --- .../com/influxdb/v3/client/internal/InfluxDBClientImpl.java | 4 +++- 1 file changed, 3 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 4ed9a571..64c1645e 100644 --- a/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java +++ b/src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java @@ -197,7 +197,9 @@ public Stream> queryRows(@Nonnull final String query) { @Nonnull @Override - public Stream> queryRows(@Nonnull final String query, @Nonnull final Map parameters) { + public Stream> queryRows(@Nonnull final String query, + @Nonnull final Map parameters + ) { return queryRows(query, parameters, QueryOptions.DEFAULTS); } From 08e5230a0ba1fe8f80f059f4796148ddda423534 Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Tue, 24 Dec 2024 17:02:33 +0700 Subject: [PATCH 04/13] feat: change Hashmap -> LinkedHashMap --- .../v3/client/internal/VectorSchemaRootConverter.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java b/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java index 4c2bb8b3..f0c00c5d 100644 --- a/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java +++ b/src/main/java/com/influxdb/v3/client/internal/VectorSchemaRootConverter.java @@ -23,7 +23,7 @@ import java.math.BigInteger; import java.time.LocalDateTime; -import java.util.HashMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -206,7 +206,7 @@ public Object[] getArrayObjectFromVectorSchemaRoot(@Nonnull final VectorSchemaRo * @return A Map represents a row of data */ public Map getMapFromVectorSchemaRoot(@Nonnull final VectorSchemaRoot vector, final int rowNumber) { - Map row = new HashMap<>(); + Map row = new LinkedHashMap<>(); for (FieldVector fieldVector : vector.getFieldVectors()) { Object mappedValue = getMappedValue( fieldVector.getField(), From ce92e79d4d6ce13b0d671a92ef788bfe249ddc17 Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Tue, 7 Jan 2025 11:40:50 +0700 Subject: [PATCH 05/13] chore: update CHANGELOG.md --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 48211487..b8ce7275 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ ## 1.1.0 [unreleased] +### Features + +1. [#209](https://github.com/InfluxCommunity/influxdb3-java/pull/209) Add function return row as a map + ## 1.0.0 [2024-12-11] ### Features From 8074f75bf4715fadf9a9a243a11fa4881538b3ba Mon Sep 17 00:00:00 2001 From: NguyenHoangSon96 Date: Thu, 16 Jan 2025 08:57:39 +0700 Subject: [PATCH 06/13] fix: queryRows comment --- .../com/influxdb/v3/client/InfluxDBClient.java | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java index c6cd89b8..69440e8d 100644 --- a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java +++ b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java @@ -187,7 +187,7 @@ Stream query(@Nonnull final String query, *

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

-     * try (Stream<Map<String, Object>> rows = client.query("select * from cpu where host=$host",;
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host",
      *                                                 Map.of("host", "server-a"), options)) {
      *      rows.forEach(row -> {
      *          // process row
@@ -195,7 +195,7 @@ Stream query(@Nonnull final String query,
      * });
      *
* - * @param query the query string to execute, cannot be null + * @param query the query string to execute, cannot be null * @return Batches of rows returned by the query */ @Nonnull @@ -206,7 +206,7 @@ Stream query(@Nonnull final String query, *

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

-     * try (Stream<Map<String, Object>> rows = client.query("select * from cpu where host=$host",;
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host",
      *                                                 Map.of("host", "server-a"), options)) {
      *      rows.forEach(row -> {
      *          // process row
@@ -214,7 +214,7 @@ Stream query(@Nonnull final String query,
      * });
      *
* - * @param query the query string to execute, cannot be null + * @param query the query string to execute, cannot be null * @param parameters query named parameters * @return Batches of rows returned by the query */ @@ -226,7 +226,7 @@ Stream query(@Nonnull final String query, *

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

-     * try (Stream<Map<String, Object>> rows = client.query("select * from cpu where host=$host",;
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host",
      *                                                 Map.of("host", "server-a"), options)) {
      *      rows.forEach(row -> {
      *          // process row
@@ -234,7 +234,7 @@ Stream query(@Nonnull final String query,
      * });
      *
* - * @param query the query string to execute, cannot be null + * @param query the query string to execute, cannot be null * @param options the options for querying data from InfluxDB * @return Batches of rows returned by the query */ @@ -246,7 +246,7 @@ Stream query(@Nonnull final String query, *

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

-     * try (Stream<Map<String, Object>> rows = client.query("select * from cpu where host=$host",;
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host",
      *                                                 Map.of("host", "server-a"), options)) {
      *      rows.forEach(row -> {
      *          // process row

From 877a02e93cc656c3c7b93cfc00086c00356f4424 Mon Sep 17 00:00:00 2001
From: NguyenHoangSon96 
Date: Thu, 16 Jan 2025 09:41:56 +0700
Subject: [PATCH 07/13] feat: add test for getMapFromVectorSchemaRoot function

---
 .../internal/VectorSchemaRootConverterTest.java | 17 ++++++++++++++++-
 .../client/internal/VectorSchemaRootUtils.java  |  2 +-
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/src/test/java/com/influxdb/v3/client/internal/VectorSchemaRootConverterTest.java b/src/test/java/com/influxdb/v3/client/internal/VectorSchemaRootConverterTest.java
index dbbd69aa..792b5692 100644
--- a/src/test/java/com/influxdb/v3/client/internal/VectorSchemaRootConverterTest.java
+++ b/src/test/java/com/influxdb/v3/client/internal/VectorSchemaRootConverterTest.java
@@ -214,7 +214,7 @@ public void testConverterWithMetaType() {
             Assertions.assertThat(diskFree.getClass()).isEqualTo(Long.class);
 
             Double temperature = (Double) pointValues.getField("temperature");
-            Assertions.assertThat(temperature).isEqualTo(100.8766f);
+            Assertions.assertThat(temperature).isEqualTo(100.8766);
             Assertions.assertThat(temperature.getClass()).isEqualTo(Double.class);
 
             String name = (String) pointValues.getField("name");
@@ -227,6 +227,21 @@ public void testConverterWithMetaType() {
         }
     }
 
+    @Test
+    void testGetMapFromVectorSchemaRoot() {
+        try (VectorSchemaRoot root = VectorSchemaRootUtils.generateVectorSchemaRoot()) {
+            Map map = VectorSchemaRootConverter.INSTANCE.getMapFromVectorSchemaRoot(root, 0);
+
+            Assertions.assertThat(map).hasSize(7);
+            Assertions.assertThat(map.get("measurement")).isEqualTo("host");
+            Assertions.assertThat(map.get("mem_total")).isEqualTo(2048L);
+            Assertions.assertThat(map.get("temperature")).isEqualTo(100.8766);
+            Assertions.assertThat(map.get("isActive")).isEqualTo(true);
+            Assertions.assertThat(map.get("name")).isEqualTo("intel");
+            Assertions.assertThat(map.get("time")).isEqualTo(BigInteger.valueOf(123_456L * 1_000_000));
+        }
+    }
+
     @Test
     void timestampWithoutMetadataAndFieldWithoutMetadata() {
         FieldType timeType = new FieldType(true, new ArrowType.Time(TimeUnit.MILLISECOND, 32), null);
diff --git a/src/test/java/com/influxdb/v3/client/internal/VectorSchemaRootUtils.java b/src/test/java/com/influxdb/v3/client/internal/VectorSchemaRootUtils.java
index 3557e037..c2510bb3 100644
--- a/src/test/java/com/influxdb/v3/client/internal/VectorSchemaRootUtils.java
+++ b/src/test/java/com/influxdb/v3/client/internal/VectorSchemaRootUtils.java
@@ -118,7 +118,7 @@ public static VectorSchemaRoot generateVectorSchemaRoot() {
 
         Float8Vector floatVector = (Float8Vector) root.getVector("temperature");
         floatVector.allocateNew();
-        floatVector.set(0, 100.8766f);
+        floatVector.set(0, 100.8766);
 
         VarCharVector stringVector = (VarCharVector) root.getVector("name");
         stringVector.allocateNew();

From a0fcaab4beacd3d3def5a014ca27a4907136a495 Mon Sep 17 00:00:00 2001
From: NguyenHoangSon96 
Date: Thu, 16 Jan 2025 15:14:50 +0700
Subject: [PATCH 08/13] feat: add more test cases for queryRows function

---
 .../v3/client/InfluxDBClientTest.java         | 118 +++++++++++++-----
 1 file changed, 88 insertions(+), 30 deletions(-)

diff --git a/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java b/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java
index 3294dcc3..8c8c8d7b 100644
--- a/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java
+++ b/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java
@@ -23,11 +23,15 @@
 
 import java.math.BigInteger;
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.Properties;
 import java.util.UUID;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import org.apache.arrow.flight.FlightRuntimeException;
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
@@ -188,45 +192,99 @@ public void testQueryRows() throws Exception {
                 System.getenv("TESTING_INFLUXDB_DATABASE"),
                 null)) {
             String uuid = UUID.randomUUID().toString();
-            long timestamp = Instant.now().getEpochSecond();
-            String record = String.format(
-                    "host12,tag=tagValue "
-                            + "name=\"intel\","
-                            + "mem_total=2048,"
-                            + "disk_free=100i,"
-                            + "temperature=100.86,"
-                            + "isActive=true,"
-                            + "testId=\"%s\" %d",
-                    uuid,
-                    timestamp
-            );
-            client.writeRecord(record, new WriteOptions(null, WritePrecision.S, null));
+            String measurement = "host21";
+            List> testDatas = new ArrayList<>();
+            for (int i = 0; i <= 9; i++) {
+                long timestamp = System.currentTimeMillis();
+                Map map = Map.of(
+                        "measurement", measurement,
+                        "tag", "tagValue",
+                        "name", "intel",
+                        "mem_total", 2048.0,
+                        "disk_free", 100L,
+                        "temperature", 100.86,
+                        "isActive", true,
+                        "time", timestamp,
+                        "testId", uuid
+                );
+                String record = String.format(
+                        "%s,tag=tagValue "
+                                + "name=\"%s\","
+                                + "mem_total=%f,"
+                                + "disk_free=%di,"
+                                + "temperature=%f,"
+                                + "isActive=%b,"
+                                + "testId=\"%s\" %d",
+                        measurement,
+                        map.get("name"),
+                        (Double) map.get("mem_total"),
+                        (Long) map.get("disk_free"),
+                        (Double) map.get("temperature"),
+                        map.get("isActive"),
+                        uuid,
+                        timestamp
+                );
+                client.writeRecord(record, new WriteOptions(null, WritePrecision.MS, null));
+                testDatas.add(map);
+            }
 
             Map parameters = Map.of("testId", uuid);
-            String sql = "Select * from host12 where \"testId\"=$testId";
+            String sql = String.format("Select * from %s where \"testId\"=$testId order by time", measurement); // Result set much be ordered by time
             try (Stream> stream = client.queryRows(sql, parameters)) {
-                stream.findFirst()
-                      .ifPresent(map -> {
-                          Assertions.assertThat(map.get("tag").getClass()).isEqualTo(String.class);
-                          Assertions.assertThat(map.get("tag")).isEqualTo("tagValue");
+                List> results = stream.collect(Collectors.toList());
+                for (int i = 0; i <= 9; i++) {
+                    Map row = results.get(i);
+                    Map testData = testDatas.get(i);
+                    Assertions.assertThat(row.get("tag").getClass()).isEqualTo(String.class);
+                    Assertions.assertThat(row.get("tag")).isEqualTo(testData.get("tag"));
 
-                          Assertions.assertThat(map.get("name").getClass()).isEqualTo(String.class);
-                          Assertions.assertThat(map.get("name")).isEqualTo("intel");
+                    Assertions.assertThat(row.get("name").getClass()).isEqualTo(String.class);
+                    Assertions.assertThat(row.get("name")).isEqualTo(testData.get("name"));
 
-                          Assertions.assertThat(map.get("mem_total").getClass()).isEqualTo(Double.class);
-                          Assertions.assertThat(map.get("mem_total")).isEqualTo(2048.0);
+                    Assertions.assertThat(row.get("mem_total").getClass()).isEqualTo(Double.class);
+                    Assertions.assertThat(row.get("mem_total")).isEqualTo(testData.get("mem_total"));
 
-                          Assertions.assertThat(map.get("disk_free").getClass()).isEqualTo(Long.class);
-                          Assertions.assertThat(map.get("disk_free")).isEqualTo(100L);
+                    Assertions.assertThat(row.get("disk_free").getClass()).isEqualTo(Long.class);
+                    Assertions.assertThat(row.get("disk_free")).isEqualTo(testData.get("disk_free"));
 
-                          Assertions.assertThat(map.get("isActive").getClass()).isEqualTo(Boolean.class);
-                          Assertions.assertThat(map.get("isActive")).isEqualTo(true);
+                    Assertions.assertThat(row.get("isActive").getClass()).isEqualTo(Boolean.class);
+                    Assertions.assertThat(row.get("isActive")).isEqualTo(testData.get("isActive"));
 
-                          Assertions.assertThat(map.get("time").getClass()).isEqualTo(BigInteger.class);
-                          Assertions.assertThat(map.get("time"))
-                                    .isEqualTo(BigInteger.valueOf(timestamp * 1_000_000_000));
-                      });
+                    Assertions.assertThat(row.get("time").getClass()).isEqualTo(BigInteger.class);
+                    Assertions.assertThat(row.get("time"))
+                              .isEqualTo(BigInteger.valueOf((Long) testData.get("time") * 1_000_000));
+                }
             }
         }
     }
+
+
+    @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_URL", matches = ".*")
+    @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_TOKEN", matches = ".*")
+    @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_DATABASE", matches = ".*")
+    @Test
+    public void testQueryRowsExceptionCases() throws Exception {
+        try (InfluxDBClient client = InfluxDBClient.getInstance(
+                System.getenv("TESTING_INFLUXDB_URL"),
+                System.getenv("TESTING_INFLUXDB_TOKEN").toCharArray(),
+                System.getenv("TESTING_INFLUXDB_DATABASE"),
+                null)) {
+
+            // Empty result case
+            Map parameters = Map.of("testId", "NotExist");
+            String sql = "Select * from host21 where \"testId\"=$testId";
+            try (Stream> stream = client.queryRows(sql, parameters)) {
+                Assertions.assertThat((int) stream.count()).isEqualTo(0);
+            }
+
+            // Malformed query case
+            Assertions.assertThatThrownBy(() -> {
+                          String query = "Select * from host21 whereabs testId=2";
+                          try (Stream> stream = client.queryRows(query)) {
+                              stream.findFirst();
+                          }
+                      })
+                      .isInstanceOf(FlightRuntimeException.class);
+        }
+    }
 }

From 5fe04410f775d4526c4f16343e78b05e04cbedd3 Mon Sep 17 00:00:00 2001
From: NguyenHoangSon96 
Date: Thu, 16 Jan 2025 15:18:51 +0700
Subject: [PATCH 09/13] fix: linter

---
 src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java b/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java
index 8c8c8d7b..f0e7ef89 100644
--- a/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java
+++ b/src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java
@@ -229,7 +229,8 @@ public void testQueryRows() throws Exception {
             }
 
             Map parameters = Map.of("testId", uuid);
-            String sql = String.format("Select * from %s where \"testId\"=$testId order by time", measurement); // Result set much be ordered by time
+            // Result set much be ordered by time
+            String sql = String.format("Select * from %s where \"testId\"=$testId order by time", measurement);
             try (Stream> stream = client.queryRows(sql, parameters)) {
                 List> results = stream.collect(Collectors.toList());
                 for (int i = 0; i <= 9; i++) {

From 5bfc37d64f470845a36625a3e7927ba9d9742321 Mon Sep 17 00:00:00 2001
From: NguyenHoangSon96 
Date: Thu, 23 Jan 2025 15:29:05 +0700
Subject: [PATCH 10/13] fix: comment

---
 .../influxdb/v3/client/InfluxDBClient.java    | 36 +++++++++----------
 1 file changed, 16 insertions(+), 20 deletions(-)

diff --git a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java
index 69440e8d..28fae150 100644
--- a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java
+++ b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java
@@ -187,12 +187,11 @@ Stream query(@Nonnull final String query,
      * 

      * The result stream should be closed after use, you can use try-resource pattern to close it automatically:
      * 
-     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host",
-     *                                                 Map.of("host", "server-a"), options)) {
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=intel")) {
      *      rows.forEach(row -> {
      *          // process row
-     *      }
-     * });
+     *      });
+     * };
      * 

      *
      * @param query the query string to execute, cannot be null
@@ -206,15 +205,14 @@ Stream query(@Nonnull final String query,
      * 

      * The result stream should be closed after use, you can use try-resource pattern to close it automatically:
      * 
-     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host",
-     *                                                 Map.of("host", "server-a"), options)) {
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host", Map.of("host", "server-a"))) {
      *      rows.forEach(row -> {
      *          // process row
-     *      }
-     * });
+     *      })
+     * };
      * 

      *
-     * @param query the query string to execute, cannot be null
+     * @param query      the query string to execute, cannot be null
      * @param parameters query named parameters
      * @return Batches of rows returned by the query
      */
@@ -226,15 +224,14 @@ Stream query(@Nonnull final String query,
      * 

      * The result stream should be closed after use, you can use try-resource pattern to close it automatically:
      * 
-     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host",
-     *                                                 Map.of("host", "server-a"), options)) {
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=intel", options)) {
      *      rows.forEach(row -> {
      *          // process row
-     *      }
-     * });
+     *      })
+     * };
      * 

      *
-     * @param query the query string to execute, cannot be null
+     * @param query   the query string to execute, cannot be null
      * @param options the options for querying data from InfluxDB
      * @return Batches of rows returned by the query
      */
@@ -246,12 +243,11 @@ Stream query(@Nonnull final String query,
      * 

      * The result stream should be closed after use, you can use try-resource pattern to close it automatically:
      * 
-     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host",
-     *                                                 Map.of("host", "server-a"), options)) {
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host", Map.of("host", "server-a"), options)) {
      *      rows.forEach(row -> {
      *          // process row
-     *      }
-     * });
+     *      })
+     * };
      * 

      *
      * @param query      the query string to execute, cannot be null
@@ -261,8 +257,8 @@ Stream query(@Nonnull final String query,
      */
     @Nonnull
     Stream> queryRows(@Nonnull final String query,
-                           @Nonnull final Map parameters,
-                           @Nonnull final QueryOptions options);
+                                          @Nonnull final Map parameters,
+                                          @Nonnull final QueryOptions options);
 
     /**
      * Query data from InfluxDB IOx into Point structure using FlightSQL.

From 865fb7d07971f1d9c2cc0b9b9d129eb0bc326dde Mon Sep 17 00:00:00 2001
From: NguyenHoangSon96 
Date: Thu, 23 Jan 2025 15:34:09 +0700
Subject: [PATCH 11/13] fix: linter

---
 src/main/java/com/influxdb/v3/client/InfluxDBClient.java | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java
index 28fae150..7a1abbfe 100644
--- a/src/main/java/com/influxdb/v3/client/InfluxDBClient.java
+++ b/src/main/java/com/influxdb/v3/client/InfluxDBClient.java
@@ -205,7 +205,8 @@ Stream query(@Nonnull final String query,
      * 

      * The result stream should be closed after use, you can use try-resource pattern to close it automatically:
      * 
-     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host", Map.of("host", "server-a"))) {
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host",
+     *                                                                      Map.of("host", "server-a"))) {
      *      rows.forEach(row -> {
      *          // process row
      *      })
@@ -224,7 +225,8 @@ Stream query(@Nonnull final String query,
      * 

      * The result stream should be closed after use, you can use try-resource pattern to close it automatically:
      * 
-     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=intel", options)) {
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=intel",
+     *                                                                      options)) {
      *      rows.forEach(row -> {
      *          // process row
      *      })
@@ -243,7 +245,8 @@ Stream query(@Nonnull final String query,
      * 

      * The result stream should be closed after use, you can use try-resource pattern to close it automatically:
      * 
-     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host", Map.of("host", "server-a"), options)) {
+     * try (Stream<Map<String, Object>> rows = client.queryRows("select * from cpu where host=$host",
+     *                                                                      Map.of("host", "server-a"), options)) {
      *      rows.forEach(row -> {
      *          // process row
      *      })

From 78addd73f3b6c81d22ed13a103374a9614b418cf Mon Sep 17 00:00:00 2001
From: NguyenHoangSon96 
Date: Tue, 11 Mar 2025 14:01:32 +0700
Subject: [PATCH 12/13] [EMPTY] trigger CI


From 2bb7cc2921c8cd5f9dde43e2910e27baf73a80ab Mon Sep 17 00:00:00 2001
From: NguyenHoangSon96 
Date: Fri, 14 Mar 2025 07:35:49 +0700
Subject: [PATCH 13/13] chore: CHANGELOG.md

---
 CHANGELOG.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b8ce7275..ca69fcd5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,7 +2,7 @@
 
 ### Features
 
-1. [#209](https://github.com/InfluxCommunity/influxdb3-java/pull/209) Add function return row as a map
+1. [#209](https://github.com/InfluxCommunity/influxdb3-java/pull/209) Add query function returning row as map
 
 ## 1.0.0 [2024-12-11]