fix: tags with commas

client/src/main/java/com/influxdb/client/internal/InfluxQLQueryApiImpl.java

@@ -120,7 +120,9 @@ static InfluxQLQueryResult readInfluxQLResult(
         // All other columns are dynamically parsed
         final int dynamicColumnsStartIndex = 2;
 
-        try (CSVParser parser = new CSVParser(reader, CSVFormat.DEFAULT.builder().setIgnoreEmptyLines(false).build())) {
+        try (CSVParser parser = new CSVParser(reader, CSVFormat.DEFAULT.builder()
+            .setIgnoreEmptyLines(false)
+            .build())) {
             for (CSVRecord csvRecord : parser) {
                 if (cancellable.isCancelled()) {
                     break;
@@ -180,11 +182,84 @@ static InfluxQLQueryResult readInfluxQLResult(
 
     private static Map<String, String> parseTags(@Nonnull final String value) {
         final Map<String, String> tags = new HashMap<>();
-        if (value.length() > 0) {
-            for (String entry : value.split(",")) {
-                final String[] kv = entry.split("=");
-                tags.put(kv[0], kv[1]);
+        if (value.isEmpty()) {
+            return tags;
+        }
+
+        StringBuilder currentKey  = new StringBuilder();
+        StringBuilder currentValue = new StringBuilder();
+        boolean inValue = false;
+        boolean escaped = false;
+        boolean firstEscaped = false;
+
+        for (int i = 0; i < value.length(); i++) {
+            char c = value.charAt(i);
+
+            if (escaped) {
+                // current character is escaped - treat it as a literal
+                if (inValue) {
+                    currentValue.append(c);
+                } else {
+                    currentKey.append(c);
+                }
+                escaped = false;
+                continue;
+            }
+
+            if (c == '\\') {
+                // start escape sequence
+                // preserve escape character
+                if (firstEscaped) {
+                    escaped = true;
+                    firstEscaped = false;
+                    continue;
+                }
+                if (inValue) {
+                    currentValue.append(c);
+                } else {
+                    currentKey.append(c);
+                }
+                firstEscaped = true;
+                continue;
             }
+
+            if(firstEscaped) {
+                firstEscaped = false;
+                continue;
+            }
+
+            if (!inValue && c == '=') {
+                // unescaped '=' marks copula
+                inValue = true;
+                continue;
+            }
+
+            if (inValue && c == ',') {
+                // unescaped comma separates key value pairs
+                // finalize
+                String key = currentKey.toString();
+                String val = currentValue.toString();
+                if (!key.isEmpty()) {
+                    tags.put(key, val);
+                }
+                currentKey.setLength(0);
+                currentValue.setLength(0);
+                inValue = false;
+                continue;
+            }
+
+            if (inValue) {
+                currentValue.append(c);
+            } else {
+                currentKey.append(c);
+            }
+        }
+
+        // finalize last key/value pair if any
+        String key =  currentKey.toString();
+        String val = currentValue.toString();
+        if (inValue && !key.isEmpty()) {
+            tags.put(key, val);
         }
 
         return tags;

client/src/test/java/com/influxdb/client/ITInfluxQLQueryApi.java

@@ -24,6 +24,8 @@
 import java.io.IOException;
 import java.math.BigDecimal;
 import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
 
 import com.influxdb.client.domain.Bucket;
 import com.influxdb.client.domain.DBRPCreate;
@@ -94,6 +96,32 @@ void testQueryData() {
 				});
 	}
 
+	@Test
+	void testQueryWithTagsWithEscapedChars() {
+		Bucket bucket = influxDBClient.getBucketsApi().findBucketByName("my-bucket");
+		influxDBClient.getWriteApiBlocking()
+			.writePoint(bucket.getId(), bucket.getOrgID(), new Point("specialTags")
+			.time(1655900000, WritePrecision.S)
+			.addField("free", 10)
+			.addTag("host", "A")
+			.addTag("region", "west")
+			.addTag("location", "vancouver\\,\\ BC")
+			.addTag("model\\,\\ uid","droid\\,\\ C3PO")
+			);
+
+		Map<String,String> expectedTags = new HashMap<>();
+		expectedTags.put("host", "A");
+		expectedTags.put("region", "west");
+		expectedTags.put("location", "vancouver\\,\\ BC");
+		expectedTags.put("model\\,\\ uid","droid\\,\\ C3PO");
+
+
+		InfluxQLQueryResult result = influxQLQueryApi.query(
+			new InfluxQLQuery("SELECT * FROM \"specialTags\" GROUP BY *", DATABASE_NAME));
+
+		Assertions.assertThat(result.getResults().get(0).getSeries().get(0).getTags()).isEqualTo(expectedTags);
+	}
+
 	@Test
 	void testQueryDataWithConversion() {
 		InfluxQLQueryResult result = influxQLQueryApi.query(

client/src/test/java/com/influxdb/client/internal/InfluxQLQueryApiImplTest.java

@@ -24,13 +24,21 @@
 import java.io.IOException;
 import java.io.StringReader;
 import java.time.Instant;
+import java.util.AbstractMap;
+import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import com.influxdb.Cancellable;
 import com.influxdb.query.InfluxQLQueryResult;
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import javax.annotation.Nonnull;
+
 class InfluxQLQueryApiImplTest {
 
 	private static final Cancellable NO_CANCELLING = new Cancellable() {
@@ -44,6 +52,143 @@ public boolean isCancelled() {
 		}
 	};
 
+	private static Map<String, String> mapOf(@Nonnull final String... valuePairs) {
+		Map<String, String> map = new HashMap<>();
+		if (valuePairs.length % 2 != 0) {
+			throw new IllegalArgumentException("value pairs must be even");
+		}
+		for (int i = 0; i < valuePairs.length; i += 2) {
+			map.put(valuePairs[i], valuePairs[i + 1]);
+		}
+		return map;
+	}
+
+	@Test
+	void readInfluxQLResultWithTagCommas() throws IOException {
+		InfluxQLQueryResult.Series.ValueExtractor extractValue = (columnName, rawValue, resultIndex, seriesName) -> {
+			if (resultIndex == 0 && seriesName.equals("data1")){
+				switch (columnName){
+					case "time": return Instant.ofEpochSecond(Long.parseLong(rawValue));
+					case "first":
+						return Double.valueOf(rawValue);
+				}
+			}
+			return rawValue;
+		};
+
+		// Note that escapes in tags returned from server are themselves escaped
+		List<String> testTags = Arrays.asList(
+			"location=Cheb_CZ", //simpleTag
+			"region=us-east-1,host=server1", // standardTags * 2
+			"location=Cheb\\\\,\\\\ CZ", // simpleTag with value comma and space
+			"location=Cheb_CZ,branch=Munchen_DE", // multiple tags with underscore
+			"location=Cheb\\\\,\\\\ CZ,branch=Munchen\\\\,\\\\ DE", // multiple tags with comma and space
+			"model\\\\,\\\\ uin=C3PO", // tag with comma space in key
+			"model\\\\,\\\\ uin=Droid\\\\, C3PO", // tag with comma space in key and value
+			"model\\\\,\\\\ uin=Droid\\\\,\\\\ C3PO,location=Cheb\\\\,\\\\ CZ,branch=Munchen\\\\,\\\\ DE", // comma space in key and val
+			"silly\\\\,\\\\=long\\\\,tag=a\\\\,b\\\\\\\\\\,\\\\ c\\\\,\\\\ d", // multi commas in k and v plus escaped reserved chars
+			"region=us\\\\,\\\\ east-1,host\\\\,\\\\ name=ser\\\\,\\\\ ver1" // legacy broken tags
+		);
+
+		Map<String,Map<String,String>> expectedTagsMap = Stream.of(
+			// 1. simpleTag
+			new AbstractMap.SimpleImmutableEntry<>(testTags.get(0),
+				mapOf("location", "Cheb_CZ")),
+			// 2. standardTags * 2
+			new AbstractMap.SimpleImmutableEntry<>(testTags.get(1),
+				mapOf(
+					"region", "us-east-1",
+					"host", "server1"
+				)),
+			// 3. simpleTag with value comma and space
+			new AbstractMap.SimpleImmutableEntry<>(testTags.get(2),
+				mapOf("location", "Cheb\\,\\ CZ")),
+			// 4. multiple tags with underscore
+			new AbstractMap.SimpleImmutableEntry<>(testTags.get(3),
+				mapOf(
+					"location", "Cheb_CZ",
+					"branch", "Munchen_DE"
+				)),
+			// 5. multiple tags with comma and space
+			new AbstractMap.SimpleImmutableEntry<>(testTags.get(4),
+				mapOf(
+					"location", "Cheb\\,\\ CZ",
+					"branch", "Munchen\\,\\ DE"
+				)),
+			// 6. tag with comma and space in key
+			new AbstractMap.SimpleImmutableEntry<>(testTags.get(5),
+				mapOf("model\\,\\ uin", "C3PO")),
+			// 7. tag with comma and space in key and value
+			new AbstractMap.SimpleImmutableEntry<>(testTags.get(6),
+				mapOf("model\\,\\ uin", "Droid\\, C3PO")),
+			// 8. comma space in key and val with multiple tags
+			new AbstractMap.SimpleImmutableEntry<>(testTags.get(7),
+				mapOf(
+					"model\\,\\ uin", "Droid\\,\\ C3PO",
+					"location", "Cheb\\,\\ CZ",
+					"branch", "Munchen\\,\\ DE"
+				)),
+			// 9. multiple commas in key and value
+		    new AbstractMap.SimpleImmutableEntry<>(testTags.get(8),
+				mapOf(
+					"silly\\,\\=long\\,tag", "a\\,b\\\\\\,\\ c\\,\\ d"
+				)),
+			// legacy broken tags
+			new AbstractMap.SimpleImmutableEntry<>(testTags.get(9),
+				mapOf(
+					"region", "us\\,\\ east-1",
+					"host\\,\\ name", "ser\\,\\ ver1"
+				))
+		).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+
+		StringReader reader = new StringReader("name,tags,time,first\n"
+			+ "data1,\"" + testTags.get(0) + "\",1483225200,42\n"
+			+ "data1,\"" + testTags.get(1) + "\",1483225200,42\n"
+			+ "data1,\"" + testTags.get(2) + "\",1483225200,42\n"
+			+ "data1,\"" + testTags.get(3) + "\",1483225200,42\n"
+			+ "data1,\"" + testTags.get(4) + "\",1483225200,42\n"
+			+ "data1,\"" + testTags.get(5) + "\",1483225200,42\n"
+			+ "data1,\"" + testTags.get(6) + "\",1483225200,42\n"
+			+ "data1,\"" + testTags.get(7) + "\",1483225200,42\n"
+			+ "data1,\"" + testTags.get(8) + "\",1483225200,42\n"
+			+ "\n"
+			+ "name,tags,time,usage_user,usage_system\n"
+			+ "cpu,\"" + testTags.get(9) + "\",1483225200,13.57,1.4\n"
+		);
+
+		InfluxQLQueryResult result = InfluxQLQueryApiImpl.readInfluxQLResult(reader, NO_CANCELLING, extractValue);
+		List<InfluxQLQueryResult.Result> results = result.getResults();
+		int index = 0;
+		for(InfluxQLQueryResult.Result r : results) {
+			for(InfluxQLQueryResult.Series s : r.getSeries()){
+				Assertions.assertThat(s.getTags()).isEqualTo(expectedTagsMap.get(testTags.get(index++)));
+				if(index < 10) {
+					Assertions.assertThat(s.getColumns()).containsOnlyKeys("time", "first");
+					InfluxQLQueryResult.Series.Record valRec = s.getValues().get(0);
+					Assertions.assertThat(valRec.getValueByKey("first")).isEqualTo(Double.valueOf("42.0"));
+					Assertions.assertThat(valRec.getValueByKey("time")).isEqualTo(Instant.ofEpochSecond(1483225200L));
+				} else if (index == 10) {
+					Assertions.assertThat(s.getColumns()).containsOnlyKeys("time", "usage_user", "usage_system");
+					InfluxQLQueryResult.Series.Record valRec = s.getValues().get(0);
+					// No value extractor created for "cpu" series
+					Assertions.assertThat(valRec.getValueByKey("time")).isEqualTo("1483225200");
+					Assertions.assertThat(valRec.getValueByKey("usage_user")).isEqualTo("13.57");
+					Assertions.assertThat(valRec.getValueByKey("usage_system")).isEqualTo("1.4");
+				}
+			}
+		}
+		Assertions.assertThat(index).isEqualTo(testTags.size());
+	}
+
+	/*
+	Sample response 1 - note escaped commas
+name,tags,time,fVal,iVal,id,location,"location\,boo",model,"model\,uin",sVal
+zaphrod_b,,1773307528202967039,26.54671,-6922649068284626682,bar,Harfa,,R2D2,,FOO
+zaphrod_b,,1773322199131651270,26.54671,-6922649068284626682,bar,,Harfa,R2D2,,FOO
+zaphrod_b,,1773322228235655514,26.54671,-6922649068284626682,bar,,"Harfa\,\ Praha",R2D2,,FOO
+zaphrod_b,,1773322254827374192,26.54671,-6922649068284626682,bar,,"Harfa\,\ Praha",,R2D2,FOO
+	 */
+
 	@Test
 	void readInfluxQLResult() throws IOException {
 		InfluxQLQueryResult.Series.ValueExtractor extractValues = (columnName, rawValue, resultIndex, seriesName) -> {