wip

Author: NguyenHoangSon96

src/main/java/com/influxdb/v3/client/internal/InfluxDBClientImpl.java

@@ -42,6 +42,7 @@
 import io.netty.handler.codec.http.HttpMethod;
 import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.util.Text;
 
 import com.influxdb.v3.client.InfluxDBApiException;
 import com.influxdb.v3.client.InfluxDBClient;
@@ -206,9 +207,9 @@ public Stream<Object[]> query(@Nonnull final String query,
                                                         row.add(doubleValue);
                                                         break;
                                                     case "iox::column_type::field::string":
-                                                        var stringValue = (String) fieldVectors.get(i)
-                                                                                               .getObject(rowNumber);
-                                                        row.add(stringValue);
+                                                        var textValue = (Text) fieldVectors.get(i)
+                                                                                           .getObject(rowNumber);
+                                                        row.add(textValue.toString());
                                                         break;
                                                     case "iox::column_type::field::boolean":
                                                         var boolValue = (Boolean) fieldVectors.get(i)
@@ -222,7 +223,9 @@ public Stream<Object[]> query(@Nonnull final String query,
                                                 BigInteger time = NanosecondConverter.getTimestamp(timestamp, schema);
                                                 row.add(time);
                                             } else {
-                                                row.add(fieldVectors.get(i).getObject(rowNumber));
+                                                Text textValue = (Text) fieldVectors.get(i)
+                                                                                    .getObject(rowNumber);
+                                                row.add(textValue.toString());
                                             }
                                         }
 

src/test/java/com/influxdb/v3/client/InfluxDBClientTest.java

@@ -21,11 +21,19 @@
  */
 package com.influxdb.v3.client;
 
+import java.math.BigInteger;
+import java.time.Instant;
 import java.util.Map;
 import java.util.Properties;
+import java.util.UUID;
+import java.util.stream.Stream;
 
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+import com.influxdb.v3.client.write.WriteOptions;
+import com.influxdb.v3.client.write.WritePrecision;
 
 class InfluxDBClientTest {
 
@@ -116,4 +124,44 @@ public void unsupportedQueryParams() throws Exception {
                             + "class com.influxdb.v3.client.internal.InfluxDBClientImpl");
         }
     }
+
+    @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_URL", matches = ".*")
+    @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_TOKEN", matches = ".*")
+    @EnabledIfEnvironmentVariable(named = "TESTING_INFLUXDB_DATABASE", matches = ".*")
+    @Test
+    public void testQuery() 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(
+                    "host9,tag=empty name=\"intel\",mem_total=2048,disk_free=100,temperature=100.86,isActive=true,testId=\"%s\" %d",
+                    uuid,
+                    timestamp
+            );
+            client.writeRecord(record, new WriteOptions(null, WritePrecision.S, null));
+
+            Map<String, Object> parameters = Map.of("testId", uuid);
+            String sql = "Select * from host9 where \"testId\"=$testId";
+            try (Stream<Object[]> stream = client.query(sql, parameters)) {
+                stream.findFirst()
+                      .ifPresent(objects -> {
+                          Assertions.assertThat(objects[0].getClass()).isEqualTo(Double.class);
+                          Assertions.assertThat(objects[0]).isEqualTo(100.0);
+
+                          Assertions.assertThat(objects[1].getClass()).isEqualTo(Boolean.class);
+                          Assertions.assertThat(objects[1]).isEqualTo(true);
+
+                          Assertions.assertThat(objects[3].getClass()).isEqualTo(String.class);
+                          Assertions.assertThat(objects[3]).isEqualTo("intel");
+
+                          Assertions.assertThat(objects[7].getClass()).isEqualTo(BigInteger.class);
+                          Assertions.assertThat(objects[7]).isEqualTo(BigInteger.valueOf(timestamp * 1_000_000_000));
+                      });
+            }
+        }
+    }
 }