This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new def6c1735bb fix: Preserve user error classification from data servers
(#20633)
def6c1735bb is described below
commit def6c1735bbcc9e34564c908c83a67de0a381886
Author: P_K <[email protected]>
AuthorDate: Wed Oct 21 18:09:40 2026 +0530
fix: Preserve user error classification from data servers (#20633)
* Fix user error propagation from data servers
* fix: Preserve DruidException propagation and existing exception behavior
---
.../epinephelinae/GroupByMergingQueryRunner.java | 3 +
.../druid/segment/projections/Projections.java | 3 +
.../groupby/GroupByQueryRunnerFailureTest.java | 49 +++++++++++++
.../druid/segment/CursorFactoryProjectionTest.java | 21 ++++++
.../apache/druid/client/JsonParserIterator.java | 28 +++++---
.../druid/client/JsonParserIteratorTest.java | 84 ++++++++++++++++++++++
.../druid/sql/calcite/CalciteSubqueryTest.java | 2 +-
7 files changed, 181 insertions(+), 9 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/GroupByMergingQueryRunner.java
b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/GroupByMergingQueryRunner.java
index 699f659cede..791629bcd2c 100644
---
a/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/GroupByMergingQueryRunner.java
+++
b/processing/src/main/java/org/apache/druid/query/groupby/epinephelinae/GroupByMergingQueryRunner.java
@@ -406,6 +406,9 @@ public class GroupByMergingQueryRunner implements
QueryRunner<ResultRow>
if (cause instanceof TimeoutException) {
throw new QueryTimeoutException(StringUtils.nonStrictFormat("Query
timeout, cancelling pending results for query [%s]. Per-segment timeout
exceeded.", query.getId()));
}
+ if (cause instanceof DruidException) {
+ throw (DruidException) cause;
+ }
throw new RuntimeException(e);
}
}
diff --git
a/processing/src/main/java/org/apache/druid/segment/projections/Projections.java
b/processing/src/main/java/org/apache/druid/segment/projections/Projections.java
index 960adbb7107..66f9bead54e 100644
---
a/processing/src/main/java/org/apache/druid/segment/projections/Projections.java
+++
b/processing/src/main/java/org/apache/druid/segment/projections/Projections.java
@@ -143,6 +143,9 @@ public class Projections
}
}
if (name != null) {
+ if (projections.stream().noneMatch(projection ->
name.equals(projection.getSchema().getName()))) {
+ throw InvalidInput.exception("Projection[%s] does not exist", name);
+ }
throw InvalidInput.exception("Projection[%s] specified, but does not
satisfy query", name);
}
if
(cursorBuildSpec.getQueryContext().getOrDefault(QueryContextParameters.FORCE_PROJECTIONS))
{
diff --git
a/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerFailureTest.java
b/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerFailureTest.java
index bb4a687bfc3..2357053e39d 100644
---
a/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerFailureTest.java
+++
b/processing/src/test/java/org/apache/druid/query/groupby/GroupByQueryRunnerFailureTest.java
@@ -26,6 +26,8 @@ import com.google.common.collect.ImmutableList;
import org.apache.druid.collections.CloseableDefaultBlockingPool;
import org.apache.druid.collections.CloseableStupidPool;
import org.apache.druid.collections.ReferenceCountingResourceHolder;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.InvalidInput;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.common.granularity.Granularities;
import org.apache.druid.java.util.common.guava.Sequences;
@@ -340,6 +342,53 @@ public class GroupByQueryRunnerFailureTest
Assertions.assertEquals("Query [test] timed out", ex.getMessage());
}
+ @Test
+ @org.junit.jupiter.api.Timeout(value = 60_000L, unit =
java.util.concurrent.TimeUnit.MILLISECONDS)
+ public void testInvalidInputExceptionOnQueryable_multiThreaded()
+ {
+ final GroupByQuery query = GroupByQuery
+ .builder()
+ .setDataSource(QueryRunnerTestHelper.DATA_SOURCE)
+ .setQuerySegmentSpec(QueryRunnerTestHelper.FIRST_TO_THIRD)
+ .setDimensions(new DefaultDimensionSpec("quality", "alias"))
+ .setAggregatorSpecs(new LongSumAggregatorFactory("rows", "rows"))
+ .setGranularity(Granularities.ALL)
+ .queryId("test-invalid-input")
+ .build();
+
+ GroupByQueryRunnerFactory factory = makeQueryRunnerFactory(
+ GroupByQueryRunnerTest.DEFAULT_MAPPER,
+ new GroupByQueryConfig()
+ {
+ @Override
+ public boolean isSingleThreaded()
+ {
+ return false;
+ }
+ }
+ );
+
+ DruidException expected = InvalidInput.exception(
+ "Projection[p] specified, but does not satisfy query"
+ );
+
+ QueryRunner<ResultRow> failingRunner = (queryPlus, responseContext) -> {
+ throw expected;
+ };
+
+ QueryRunner<ResultRow> mergeRunners = factory.mergeRunners(
+ processingPool,
+ List.of(runner, failingRunner)
+ );
+
+ DruidException actual = Assertions.assertThrows(
+ DruidException.class,
+ () -> GroupByQueryRunnerTestHelper.runQuery(factory, mergeRunners,
query)
+ );
+
+ Assertions.assertSame(expected, actual);
+ }
+
@Test
@org.junit.jupiter.api.Timeout(value = 60_000L, unit =
java.util.concurrent.TimeUnit.MILLISECONDS)
public void testTimeoutExceptionOnQueryable_multiThreaded()
diff --git
a/processing/src/test/java/org/apache/druid/segment/CursorFactoryProjectionTest.java
b/processing/src/test/java/org/apache/druid/segment/CursorFactoryProjectionTest.java
index ae3d8784a6f..ade6fdd3473 100644
---
a/processing/src/test/java/org/apache/druid/segment/CursorFactoryProjectionTest.java
+++
b/processing/src/test/java/org/apache/druid/segment/CursorFactoryProjectionTest.java
@@ -1448,6 +1448,27 @@ public class CursorFactoryProjectionTest extends
InitializedNullHandlingTest
);
}
+ @Test
+ public void testTimeseriesQueryMissingNamedProjection()
+ {
+ final TimeseriesQuery query = Druids.newTimeseriesQueryBuilder()
+ .dataSource("test")
+
.intervals(ImmutableList.of(Intervals.ETERNITY))
+ .granularity(Granularities.DAY)
+ .aggregators(new
LongSumAggregatorFactory("c_sum", "c"))
+
.context(QueryContext.ofMap(QueryContextParameters.USE_PROJECTION,
"does_not_exist"))
+ .build();
+
+ final CursorBuildSpec buildSpec =
TimeseriesQueryEngine.makeCursorBuildSpec(query, null);
+ DruidException e = Assertions.assertThrows(
+ DruidException.class,
+ () -> projectionsCursorFactory.makeCursorHolder(buildSpec)
+ );
+ Assertions.assertEquals(DruidException.Persona.USER, e.getTargetPersona());
+ Assertions.assertEquals(DruidException.Category.INVALID_INPUT,
e.getCategory());
+ Assertions.assertEquals("Projection[does_not_exist] does not exist",
e.getMessage());
+ }
+
@Test
public void testTimeseriesQueryOrderByNotCompatibleWithProjection()
{
diff --git
a/server/src/main/java/org/apache/druid/client/JsonParserIterator.java
b/server/src/main/java/org/apache/druid/client/JsonParserIterator.java
index b590ca00ba4..362b56dae8e 100644
--- a/server/src/main/java/org/apache/druid/client/JsonParserIterator.java
+++ b/server/src/main/java/org/apache/druid/client/JsonParserIterator.java
@@ -22,8 +22,10 @@ package org.apache.druid.client;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonToken;
import com.fasterxml.jackson.core.ObjectCodec;
+import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.error.ErrorResponse;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.logger.Logger;
@@ -41,6 +43,7 @@ import org.apache.druid.utils.CloseableUtils;
import javax.annotation.Nullable;
import java.io.IOException;
import java.io.InputStream;
+import java.util.Map;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -202,7 +205,7 @@ public class JsonParserIterator<T> implements
CloseableIterator<T>
readNextToken();
objectCodec = jp.getCodec();
} else if (nextToken == JsonToken.START_OBJECT) {
- throw convertException(readStructuredError());
+ throw readStructuredError();
} else {
String errMsg = jp.getValueAsString();
if (errMsg != null) {
@@ -249,16 +252,25 @@ public class JsonParserIterator<T> implements
CloseableIterator<T>
}
/**
- * Deserializes the structured error body {@link #init()} found in place of
a result array, with the same
- * unwrapped-{@link QueryException} handling as {@link #createParser}: the
body can span chunks, so a later-chunk
- * failure can surface from inside this read rather than from the
deserialized value. An exception that comes off
- * the stream is converted here and thrown; a QueryException that was
successfully deserialized is returned for the
- * caller to convert, so neither path is converted twice.
+ * Reads a structured error from a data server. Modern DruidException
responses must retain their original
+ * persona, category, and error code. Legacy QueryException responses still
use the existing conversion logic.
+ *
+ * A QueryException can also arrive directly from the transport while a
response body is being read; normalize
+ * those exceptions to this iterator's host, as in the other parser methods.
*/
- private QueryException readStructuredError() throws IOException
+ private RuntimeException readStructuredError() throws IOException
{
try {
- return jp.getCodec().readValue(jp, QueryException.class);
+ final Map<String, Object> error = jp.getCodec().readValue(
+ jp,
+ new TypeReference<Map<String, Object>>()
+ {
+ }
+ );
+ if ("druidException".equals(error.get("error"))) {
+ return ErrorResponse.fromMap(error).getUnderlyingException();
+ }
+ return convertException(objectMapper.convertValue(error,
QueryException.class));
}
catch (QueryException e) {
throw convertException(e);
diff --git
a/server/src/test/java/org/apache/druid/client/JsonParserIteratorTest.java
b/server/src/test/java/org/apache/druid/client/JsonParserIteratorTest.java
index f454fe6f5bf..cc46b2a695e 100644
--- a/server/src/test/java/org/apache/druid/client/JsonParserIteratorTest.java
+++ b/server/src/test/java/org/apache/druid/client/JsonParserIteratorTest.java
@@ -22,9 +22,13 @@ package org.apache.druid.client;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.smile.SmileFactory;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.AbstractFuture;
import com.google.common.util.concurrent.Futures;
+import org.apache.druid.error.DruidException;
+import org.apache.druid.error.ErrorResponse;
+import org.apache.druid.error.InvalidInput;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
@@ -47,6 +51,7 @@ import org.mockito.Mockito;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
@@ -58,6 +63,85 @@ public class JsonParserIteratorTest
private static final String HOST = "host";
private static final ObjectMapper OBJECT_MAPPER = new DefaultObjectMapper();
+ @Test
+ public void testModernDruidExceptionJsonResponse() throws IOException
+ {
+ assertModernErrorResponse(OBJECT_MAPPER);
+ }
+
+ @Test
+ public void testModernDruidExceptionSmileResponse() throws IOException
+ {
+ assertModernErrorResponse(new ObjectMapper(new SmileFactory()));
+ }
+
+ @Test
+ public void testLegacyQueryTimeoutSmileResponse() throws IOException
+ {
+ final ObjectMapper mapper = new ObjectMapper(new SmileFactory());
+ final JsonParserIterator<Object> iterator = new JsonParserIterator<>(
+ JAVA_TYPE,
+ Futures.immediateFuture(new
ByteArrayInputStream(mapper.writeValueAsBytes(
+ Map.of(
+ "error", QueryException.QUERY_TIMEOUT_ERROR_CODE,
+ "errorMessage", "legacy timeout",
+ "errorClass", QueryTimeoutException.class.getName(),
+ "host", "data-server-host"
+ )
+ ))),
+ URL,
+ null,
+ HOST,
+ mapper
+ );
+
+ QueryTimeoutException actual =
Assertions.assertThrows(QueryTimeoutException.class, iterator::hasNext);
+ Assertions.assertEquals("legacy timeout", actual.getMessage());
+ Assertions.assertEquals(HOST, actual.getHost());
+ }
+
+ @Test
+ public void testMalformedModernErrorResponseIsDefensive() throws IOException
+ {
+ final JsonParserIterator<Object> iterator = new JsonParserIterator<>(
+ JAVA_TYPE,
+ Futures.immediateFuture(new
ByteArrayInputStream(OBJECT_MAPPER.writeValueAsBytes(
+ Map.of("error", "druidException", "errorMessage", "missing
persona, category, and error code")
+ ))),
+ URL,
+ null,
+ HOST,
+ OBJECT_MAPPER
+ );
+
+ DruidException actual = Assertions.assertThrows(DruidException.class,
iterator::hasNext);
+ Assertions.assertEquals(DruidException.Category.DEFENSIVE,
actual.getCategory());
+ }
+
+ private static void assertModernErrorResponse(ObjectMapper mapper) throws
IOException
+ {
+ DruidException expected = InvalidInput.exception("Projection[p] specified,
but does not satisfy query");
+ expected.withContext("projection", "p");
+
+ JsonParserIterator<Object> iterator = new JsonParserIterator<>(
+ JAVA_TYPE,
+ Futures.immediateFuture(new ByteArrayInputStream(
+ mapper.writeValueAsBytes(new ErrorResponse(expected).getAsMap())
+ )),
+ URL,
+ null,
+ HOST,
+ mapper
+ );
+
+ DruidException actual = Assertions.assertThrows(DruidException.class,
iterator::hasNext);
+ Assertions.assertEquals(expected.getErrorCode(), actual.getErrorCode());
+ Assertions.assertEquals(DruidException.Persona.USER,
actual.getTargetPersona());
+ Assertions.assertEquals(DruidException.Category.INVALID_INPUT,
actual.getCategory());
+ Assertions.assertEquals(expected.getMessage(), actual.getMessage());
+ Assertions.assertEquals("p", actual.getContext().get("projection"));
+ }
+
@SuppressWarnings("ResultOfMethodCallIgnored")
public static class FutureExceptionTest
{
diff --git
a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteSubqueryTest.java
b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteSubqueryTest.java
index 256a794fb78..61ab7f3bbed 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteSubqueryTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteSubqueryTest.java
@@ -1317,7 +1317,7 @@ public class CalciteSubqueryTest extends
BaseCalciteQueryTest
testQueryThrows(
"SELECT count(*) FROM wikipedia where channel = (select channel from
wikipedia order by __time desc LIMIT 2 OFFSET 6)",
RuntimeException.class,
- "java.util.concurrent.ExecutionException:
org.apache.druid.error.DruidException: Subquery expression returned more than
one row"
+ "Subquery expression returned more than one row"
);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
