feat: CSV parser is able to parse export from UI

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ### Features
 1. [#172](https://github.com/influxdata/influxdb-client-java/pull/172): flux-dsl: added `to` function without `org` parameter
+1. [#183](https://github.com/influxdata/influxdb-client-java/pull/183): CSV parser is able to parse export from UI
 
 ### Bug Fixes
 1. [#173](https://github.com/influxdata/influxdb-client-java/pull/173): Query error could be after _success_ table

client-core/src/main/java/com/influxdb/query/internal/FluxCsvParser.java

@@ -28,7 +28,9 @@
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Base64;
+import java.util.Collections;
 import java.util.List;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -53,6 +55,12 @@
  */
 public class FluxCsvParser {
 
+    private static final String ANNOTATION_DATATYPE = "#datatype";
+    private static final String ANNOTATION_GROUP = "#group";
+    private static final String ANNOTATION_DEFAULT = "#default";
+    private static final List<String> ANNOTATIONS = Arrays
+            .asList(ANNOTATION_DATATYPE, ANNOTATION_GROUP, ANNOTATION_DEFAULT);
+
     private enum ParsingState {
         NORMAL,
 
@@ -125,6 +133,7 @@ public void parseFluxResponse(@Nonnull final BufferedSource bufferedSource,
             int tableId = -1;
             boolean startNewTable = false;
             FluxTable table = null;
+            List<String> groups = Collections.emptyList();
             for (CSVRecord csvRecord : parser) {
 
                 if (cancellable.isCancelled()) {
@@ -157,10 +166,11 @@ public void parseFluxResponse(@Nonnull final BufferedSource bufferedSource,
 
                 String token = csvRecord.get(0);
                 //// start new table
-                if ("#datatype".equals(token)) {
+                if (ANNOTATIONS.contains(token) && !startNewTable) {
                     startNewTable = true;
 
                     table = new FluxTable();
+                    groups = Collections.emptyList();
                     consumer.accept(tableIndex, cancellable, table);
                     tableIndex++;
                     tableId = -1;
@@ -171,18 +181,17 @@ public void parseFluxResponse(@Nonnull final BufferedSource bufferedSource,
                 }
 
                 //#datatype,string,long,dateTime:RFC3339,dateTime:RFC3339,dateTime:RFC3339,double,string,string,string
-                if ("#datatype".equals(token)) {
+                if (ANNOTATION_DATATYPE.equals(token)) {
                     addDataTypes(table, toList(csvRecord));
 
-                } else if ("#group".equals(token)) {
-                    addGroups(table, toList(csvRecord));
-
-                } else if ("#default".equals(token)) {
+                } else if (ANNOTATION_GROUP.equals(token)) {
+                    groups = toList(csvRecord);
+                } else if (ANNOTATION_DEFAULT.equals(token)) {
                     addDefaultEmptyValues(table, toList(csvRecord));
-
                 } else {
                     // parse column names
                     if (startNewTable) {
+                        addGroups(table, groups);
                         addColumnNamesAndTags(table, toList(csvRecord));
                         startNewTable = false;
                         continue;
@@ -297,12 +306,12 @@ private void addGroups(@Nonnull final FluxTable table, @Nonnull final List<Strin
         Arguments.checkNotNull(table, "table");
         Arguments.checkNotNull(groups, "groups");
 
-        for (int i = 0; i < groups.size(); i++) {
+        for (int i = 0; i < table.getColumns().size(); i++) {
 
             FluxColumn fluxColumn = getFluxColumn(i, table);
 
             String group = groups.get(i);
-            fluxColumn.setGroup(Boolean.valueOf(group));
+            fluxColumn.setGroup(Boolean.parseBoolean(group));
         }
     }
 

client-core/src/test/java/com/influxdb/query/internal/FluxCsvParserTest.java

@@ -270,6 +270,7 @@ void mappingBase64Binary() throws IOException {
         List<FluxTable> tables = parseFluxResponse(data);
 
         byte[] value = (byte[]) tables.get(0).getRecords().get(0).getValueByKey("value");
+        Assertions.assertThat(value).isNotNull();
         Assertions.assertThat(value).isNotEmpty();
         Assertions.assertThat(new String(value, UTF_8)).isEqualTo(binaryData);
 
@@ -595,6 +596,24 @@ public void responseWithError() {
                 .matches((Predicate<Throwable>) throwable -> ((FluxQueryException) throwable).reference() == 0);
     }
 
+    @Test
+    public void parseExportFromUserInterface() throws IOException {
+            String data = "#group,false,false,true,true,true,true,true,true,false,false\n"
+            + "#datatype,string,long,dateTime:RFC3339,dateTime:RFC3339,string,string,string,string,double,dateTime:RFC3339\n"
+            + "#default,mean,,,,,,,,,\n"
+            + ",result,table,_start,_stop,_field,_measurement,city,location,_value,_time\n"
+            + ",,0,1754-06-26T11:30:27.613654848Z,2040-10-27T12:13:46.485Z,temperatureC,weather,London,us-midwest,30,1975-09-01T16:59:54.5Z\n"
+            + ",,1,1754-06-26T11:30:27.613654848Z,2040-10-27T12:13:46.485Z,temperatureF,weather,London,us-midwest,86,1975-09-01T16:59:54.5Z\n";
+
+        List<FluxTable> tables = parseFluxResponse(data);
+        Assertions.assertThat(tables).hasSize(2);
+        Assertions.assertThat(tables.get(0).getRecords()).hasSize(1);
+        Assertions.assertThat(tables.get(0).getColumns().get(0).isGroup()).isFalse();
+        Assertions.assertThat(tables.get(0).getColumns().get(1).isGroup()).isFalse();
+        Assertions.assertThat(tables.get(0).getColumns().get(2).isGroup()).isTrue();
+        Assertions.assertThat(tables.get(1).getRecords()).hasSize(1);
+    }
+
     @Nonnull
     private List<FluxTable> parseFluxResponse(@Nonnull final String data) throws IOException {