wip

Author: NguyenHoangSon96

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

@@ -23,6 +23,7 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.math.BigInteger;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -183,16 +184,45 @@ public Stream<Object[]> query(@Nonnull final String query,
         return queryData(query, parameters, options)
                 .flatMap(vector -> {
                     List<FieldVector> fieldVectors = vector.getFieldVectors();
-                    return IntStream
-                            .range(0, vector.getRowCount())
-                            .mapToObj(rowNumber -> {
-
-                                ArrayList<Object> row = new ArrayList<>();
-                                for (FieldVector fieldVector : fieldVectors) {
-                                    row.add(fieldVector.getObject(rowNumber));
-                                }
-                                return row.toArray();
-                            });
+                    return IntStream.range(0, vector.getRowCount())
+                                    .mapToObj(rowNumber -> {
+                                        ArrayList<Object> row = new ArrayList<>();
+                                        for (int i = 0; i < fieldVectors.size(); i++) {
+                                            var schema = vector.getSchema().getFields().get(i);
+                                            var metaType = schema.getMetadata().get("iox::column::type");
+                                            String valueType = metaType.split("::")[2];
+                                            if ("field".equals(valueType)) {
+                                                switch (metaType) {
+                                                    case "iox::column_type::field::integer":
+                                                    case "iox::column_type::field::uinteger":
+                                                        var intValue = (Long) fieldVectors.get(i).getObject(rowNumber);
+                                                        row.add(intValue);
+                                                        break;
+                                                    case "iox::column_type::field::float":
+                                                        var doubleValue = (Double) fieldVectors.get(i).getObject(rowNumber);
+                                                        row.add(doubleValue);
+                                                        break;
+                                                    case "iox::column_type::field::string":
+                                                        var stringValue = (String) fieldVectors.get(i).getObject(rowNumber);
+                                                        row.add(stringValue);
+                                                        break;
+                                                    case "iox::column_type::field::boolean":
+                                                        var boolValue = (Boolean) fieldVectors.get(i).getObject(rowNumber);
+                                                        row.add(boolValue);
+                                                        break;
+                                                    default:
+                                                }
+                                            } else if ("timestamp".equals(valueType)) {
+                                                var timestamp = fieldVectors.get(i).getObject(rowNumber);
+                                                BigInteger time = NanosecondConverter.getTimestamp(timestamp, schema);
+                                                row.add(time);
+                                            } else {
+                                                row.add(fieldVectors.get(i).getObject(rowNumber));
+                                            }
+                                        }
+
+                                        return row.toArray();
+                                    });
                 });
     }
 

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

@@ -24,12 +24,18 @@
 import java.math.BigDecimal;
 import java.math.BigInteger;
 import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
+import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
 
 import com.influxdb.v3.client.write.WritePrecision;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
 
 import static java.util.function.Function.identity;
 
@@ -111,4 +117,40 @@ public static BigInteger convert(final Instant instant, final WritePrecision pre
 
         return FROM_NANOS.get(precision).apply(nanos);
     }
+
+    public static BigInteger getTimestamp(@Nonnull final Object value, @Nonnull final Field schema) {
+        BigInteger result = null;
+
+        if (value instanceof Long) {
+            if (schema.getFieldType().getType() instanceof ArrowType.Timestamp) {
+                ArrowType.Timestamp type = (ArrowType.Timestamp) schema.getFieldType().getType();
+                TimeUnit timeUnit;
+                switch (type.getUnit()) {
+                    case SECOND:
+                        timeUnit = TimeUnit.SECONDS;
+                        break;
+                    case MILLISECOND:
+                        timeUnit = TimeUnit.MILLISECONDS;
+                        break;
+                    case MICROSECOND:
+                        timeUnit = TimeUnit.MICROSECONDS;
+                        break;
+                    default:
+                    case NANOSECOND:
+                        timeUnit = TimeUnit.NANOSECONDS;
+                        break;
+                }
+                long nanoseconds = TimeUnit.NANOSECONDS.convert((Long) value, timeUnit);
+                BigInteger convertedTime = NanosecondConverter.convert(Instant.ofEpochSecond(0, nanoseconds), WritePrecision.NS);
+                result = NanosecondConverter.convertToNanos(convertedTime, WritePrecision.NS);
+            } else {
+                BigInteger convertedTime = NanosecondConverter.convert(Instant.ofEpochMilli((Long) value), WritePrecision.NS);
+                result = NanosecondConverter.convertToNanos(convertedTime, WritePrecision.NS);
+            }
+        } else if (value instanceof LocalDateTime) {
+            BigInteger convertedTime = NanosecondConverter.convert(((LocalDateTime) value).toInstant(ZoneOffset.UTC), WritePrecision.NS);
+            result = NanosecondConverter.convertToNanos(convertedTime, WritePrecision.NS);
+        }
+        return result;
+    }
 }