chore: explicitly close the vector after finish the processing

Author: bednar

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

@@ -30,6 +30,7 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiFunction;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -42,7 +43,6 @@
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import org.apache.arrow.flight.CallOption;
-import org.apache.arrow.vector.FieldVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
 
 import com.influxdb.v3.client.InfluxDBApiException;
@@ -187,14 +187,8 @@ public Stream<Object[]> query(@Nonnull final String query, @Nonnull final Map<St
     public Stream<Object[]> query(@Nonnull final String query,
                                   @Nonnull final Map<String, Object> parameters,
                                   @Nonnull final QueryOptions options) {
-        return queryData(query, parameters, options)
-                .flatMap(vector -> IntStream.range(0, vector.getRowCount())
-                                        .mapToObj(rowNumber ->
-                                        VectorSchemaRootConverter.INSTANCE
-                                                                 .getArrayObjectFromVectorSchemaRoot(
-                                                                         vector,
-                                                                         rowNumber
-                                                                 )));
+        return queryDataAndProcess(query, parameters, options,
+                VectorSchemaRootConverter.INSTANCE::getArrayObjectFromVectorSchemaRoot);
     }
 
     @Nonnull
@@ -222,14 +216,7 @@ public Stream<Map<String, Object>> queryRows(@Nonnull final String query, @Nonnu
     public Stream<Map<String, Object>> queryRows(@Nonnull final String query,
                                                  @Nonnull final Map<String, Object> parameters,
                                                  @Nonnull final QueryOptions options) {
-        return queryData(query, parameters, options)
-                .flatMap(vector -> IntStream.range(0, vector.getRowCount())
-                                            .mapToObj(rowNumber ->
-                                                              VectorSchemaRootConverter.INSTANCE
-                                                                      .getMapFromVectorSchemaRoot(
-                                                                              vector,
-                                                                              rowNumber
-                                                                      )));
+        return queryDataAndProcess(query, parameters, options, VectorSchemaRootConverter.INSTANCE::getMapFromVectorSchemaRoot);
     }
 
     @Nonnull
@@ -255,14 +242,8 @@ public Stream<PointValues> queryPoints(@Nonnull final String query, @Nonnull fin
     public Stream<PointValues> queryPoints(@Nonnull final String query,
                                            @Nonnull final Map<String, Object> parameters,
                                            @Nonnull final QueryOptions options) {
-        return queryData(query, parameters, options)
-                .flatMap(vector -> {
-                    List<FieldVector> fieldVectors = vector.getFieldVectors();
-                    return IntStream
-                            .range(0, vector.getRowCount())
-                            .mapToObj(row ->
-                                    VectorSchemaRootConverter.INSTANCE.toPointValues(row, fieldVectors));
-                });
+        return queryDataAndProcess(query, parameters, options,
+                (vector, rowNumber) -> VectorSchemaRootConverter.INSTANCE.toPointValues(rowNumber, vector.getFieldVectors()));
     }
 
     @Nonnull
@@ -389,6 +370,19 @@ private <T> void writeData(@Nonnull final List<T> data, @Nonnull final WriteOpti
         }
     }
 
+    @Nonnull
+    private <T> Stream<T> queryDataAndProcess(@Nonnull final String query,
+                                              @Nonnull final Map<String, Object> parameters,
+                                              @Nonnull final QueryOptions options,
+                                              @Nonnull final BiFunction<VectorSchemaRoot, Integer, T> processor) {
+        return queryData(query, parameters, options)
+                .flatMap(vector ->
+                        IntStream.range(0, vector.getRowCount())
+                                .mapToObj(rowNumber -> processor.apply(vector, rowNumber))
+                                .onClose(vector::close)
+                );
+    }
+
     @Nonnull
     private Stream<VectorSchemaRoot> queryData(@Nonnull final String query,
                                                @Nonnull final Map<String, Object> parameters,