This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/skywalking.git
The following commit(s) were added to refs/heads/master by this push:
new be0383a4fd Always send an explicit LIMIT on BanyanDB queries (#13961)
be0383a4fd is described below
commit be0383a4fd87111a781175f2633c97c2fe9cca99
Author: 吴晟 Wu Sheng <[email protected]>
AuthorDate: Wed Jul 29 07:51:39 2026 +0800
Always send an explicit LIMIT on BanyanDB queries (#13961)
BanyanDB applies its own default limit to any query that carries none --
100 rows for measures, 20 for streams/traces -- and applies it after
GROUP BY, so an over-long result set is silently truncated rather than
rejected.
OAP never sent a limit on several read paths, so an entity-scoped metrics
query returned at most 100 data points regardless of the requested range:
a 4-hour minute-step read rendered only its first 100 minutes and the rest
showed as empty, even though DurationUtils.MAX_TIME_RANGE allows up to 500
steps. The same cap silently shortened topology relation maps, instance
and process metadata lists, profiling thread snapshots and eBPF task lists.
Every BydbQL query now leaves OAP with an explicit LIMIT:
- the entity-scoped metrics read sends the exact number of assembled
duration points, the same row set the ES/JDBC DAOs fetch by explicit id;
- ad-hoc SELECT TOP sends its own N;
- anything that does not paginate itself falls back to the configured
resultWindowMaxSize (default 10000) via Conditions#limitIfAbsent, applied
in the stream/measure/trace query helpers that every DAO funnels through.
The fallback is spliced in at the start of the pagination tail rather than
appended, so it lands ahead of an OFFSET that was set first and keeps
WITH QUERY_TRACE positioned as the grammar requires.
ES and JDBC storage were never affected -- both fetch metrics rows by
explicit document id.
---
docs/en/changes/changes.md | 1 +
docs/en/setup/backend/storages/banyandb.md | 2 +
.../server-starter/src/main/resources/bydb.yml | 2 +
.../plugin/banyandb/BanyanDBStorageClient.java | 11 ++
.../plugin/banyandb/BanyanDBStorageConfig.java | 6 +
.../banyandb/measure/BanyanDBMetricsQueryDAO.java | 24 ++--
.../banyandb/stream/AbstractBanyanDBDAO.java | 8 ++
.../storage/plugin/banyandb/stream/Conditions.java | 54 +++++++-
.../plugin/banyandb/stream/ConditionsTest.java | 68 ++++++++++
.../plugin/banyandb/stream/QueryLimitTest.java | 137 +++++++++++++++++++++
10 files changed, 297 insertions(+), 16 deletions(-)
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index e90572a995..a3e4023517 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -337,6 +337,7 @@
* Surface the effective BanyanDB configuration (`bydb.yml` / `bydb-topn.yml`)
in the `/debugging/config/dump` admin API. Because the BanyanDB config moved to
a separate file in 10.2.0, a BanyanDB deployment previously showed an empty
`storage.banyandb` block in the dump; its post-environment-resolution values
are now merged into the same response under `storage.banyandb.*` (TopN rules
under `storage.banyandb.topN.*`), masked by the same secret-keyword list, via a
generic `ConfigDumpExten [...]
* Fix: an MQE `top_n(metric, N, order, attrX='value')` query whose attribute
is not a column of the target metric now returns a descriptive MQE error
instead of a raw storage `IOException` surfaced as `Internal IO exception,
query metrics error.`. Attribute columns (`attr0..attrN`) exist only on
decorated metrics (`service_*` / `endpoint_*` / `kubernetes_service_*`, set to
the layer name via OAL `.decorator(...)`) and the MAL meter base; metrics such
as relations or database / cache / mq [...]
* Migrate all BanyanDB storage read queries from the typed query-builder API
to BydbQL.
+* Fix: BanyanDB queries no longer silently truncate at the storage engine's
implicit row cap. BanyanDB applies its own default limit to any query that
carries none — 100 rows for measures, 20 for streams/traces — and applies it
*after* `GROUP BY`, so an over-long result set is cut short rather than
rejected. OAP never sent a limit on several read paths, so a metrics query
returned at most 100 data points regardless of the requested range: a 4-hour
minute-step read rendered only its first [...]
* Route LAL rules within a layer by their input type, so a single layer can
host rules over different proto inputs. Each compiled rule now carries its
effective input type (the proto class its `parsed.*` getters cast to, or `null`
for parser-based / untyped rules), and `LogFilterListener` skips any rule whose
type doesn't match the incoming log instead of running every rule in the layer.
This fixes a latent `ClassCastException` (caught and logged per log) that fired
whenever a `MESH` log [...]
#### UI
diff --git a/docs/en/setup/backend/storages/banyandb.md
b/docs/en/setup/backend/storages/banyandb.md
index bd9ecdc2dc..de9b0045af 100644
--- a/docs/en/setup/backend/storages/banyandb.md
+++ b/docs/en/setup/backend/storages/banyandb.md
@@ -46,6 +46,8 @@ global:
# A higher value can improve write performance but also increases CPU usage
on both OAP and BanyanDB Server.
concurrentWriteThreads: ${SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS:15}
# The maximum size of the dataset when the OAP loads cache, such as network
aliases.
+ # Also the row cap sent for any query that has no limit of its own, so that
a query never falls back to
+ # BanyanDB's own default (100 rows for measures, 20 for streams/traces),
which truncates results silently.
resultWindowMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_WINDOW_SIZE:10000}
# The maximum size of metadata per query.
metadataQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_SIZE:10000}
diff --git a/oap-server/server-starter/src/main/resources/bydb.yml
b/oap-server/server-starter/src/main/resources/bydb.yml
index 73f6479c66..7ec4764c23 100644
--- a/oap-server/server-starter/src/main/resources/bydb.yml
+++ b/oap-server/server-starter/src/main/resources/bydb.yml
@@ -33,6 +33,8 @@ global:
# A higher value can improve write performance but also increases CPU usage
on both OAP and BanyanDB Server.
concurrentWriteThreads: ${SW_STORAGE_BANYANDB_CONCURRENT_WRITE_THREADS:15}
# The maximum size of the dataset when the OAP loads cache, such as network
aliases.
+ # Also the row cap sent for any query that has no limit of its own, so that
a query never falls back to
+ # BanyanDB's own default (100 rows for measures, 20 for streams/traces),
which truncates results silently.
resultWindowMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_WINDOW_SIZE:10000}
# The maximum size of metadata per query.
metadataQueryMaxSize: ${SW_STORAGE_BANYANDB_QUERY_MAX_SIZE:10000}
diff --git
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java
index e0917ddb4c..8add7afd67 100644
---
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java
+++
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageClient.java
@@ -27,6 +27,7 @@ import java.util.Collections;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
+import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
@@ -79,6 +80,15 @@ public class BanyanDBStorageClient implements Client,
HealthCheckable {
final BanyanDBClient client;
private final DelegatedHealthChecker healthChecker = new
DelegatedHealthChecker();
private final int flushTimeout;
+ /**
+ * Row cap sent as the fallback {@code LIMIT} on any query that does not
paginate itself, so a query never
+ * inherits BanyanDB's much smaller server-side default. Read by
+ * {@link
org.apache.skywalking.oap.server.storage.plugin.banyandb.stream.AbstractBanyanDBDAO},
which every
+ * query DAO extends — it lives here rather than in each DAO to avoid
threading the config through ~30
+ * DAO constructors.
+ */
+ @Getter
+ private final int resultWindowMaxSize;
private final ModuleManager moduleManager;
private final Options options;
private BanyandbDatabase database;
@@ -105,6 +115,7 @@ public class BanyanDBStorageClient implements Client,
HealthCheckable {
}
this.client = new BanyanDBClient(config.getGlobal().getTargets(),
options);
this.flushTimeout = config.getGlobal().getFlushTimeout();
+ this.resultWindowMaxSize = config.getGlobal().getResultWindowMaxSize();
this.options = options;
this.moduleManager = moduleManager;
this.compatibleServerApiVersions =
config.getGlobal().getCompatibleServerApiVersions();
diff --git
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
index 1475fb8c19..b29a123521 100644
---
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
+++
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBStorageConfig.java
@@ -101,6 +101,12 @@ public class BanyanDBStorageConfig extends ModuleConfig {
*/
private int pprofTaskQueryMaxSize;
+ /**
+ * Row cap for a query that does not carry a limit of its own — cache
loads such as network aliases,
+ * and every read whose result size is not bounded by paging. It is
always sent to the server: a
+ * BydbQL query with no {@code LIMIT} would otherwise fall back to
BanyanDB's own default (100 rows
+ * for measures, 20 for streams/traces), which truncates the result
set silently.
+ */
private int resultWindowMaxSize = 10000;
private int metadataQueryMaxSize = 5000;
private int segmentQueryMaxSize = 200;
diff --git
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
index bf01866699..1181a3243d 100644
---
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
+++
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/measure/BanyanDBMetricsQueryDAO.java
@@ -67,9 +67,8 @@ public class BanyanDBMetricsQueryDAO extends
AbstractBanyanDBDAO implements IMet
}
final String entityID = condition.getEntity().buildId();
- Map<Long, DataPoint> idMap = queryByEntityID(schema, valueColumnName,
duration, entityID);
-
List<PointOfTime> tsPoints = duration.assembleDurationPoints();
+ Map<Long, DataPoint> idMap = queryByEntityID(schema, valueColumnName,
duration, entityID, tsPoints.size());
MetricsValues metricsValues = new MetricsValues();
// Label is null, because in readMetricsValues, no label parameter.
@@ -105,9 +104,9 @@ public class BanyanDBMetricsQueryDAO extends
AbstractBanyanDBDAO implements IMet
@Override
public List<MetricsValues> readLabeledMetricsValues(MetricsCondition
condition, String valueColumnName, List<KeyValue> labels, Duration duration)
throws IOException {
- Map<Long, DataPoint> idMap = queryByEntityID(condition,
valueColumnName, duration);
-
List<PointOfTime> tsPoints = duration.assembleDurationPoints();
+ Map<Long, DataPoint> idMap = queryByEntityID(condition,
valueColumnName, duration, tsPoints.size());
+
String entityID = condition.getEntity().buildId();
List<String> ids = new ArrayList<>(tsPoints.size());
@@ -214,14 +213,14 @@ public class BanyanDBMetricsQueryDAO extends
AbstractBanyanDBDAO implements IMet
@Override
public HeatMap readHeatMap(MetricsCondition condition, String
valueColumnName, Duration duration) throws IOException {
- Map<Long, DataPoint> idMap = queryByEntityID(condition,
valueColumnName, duration);
+ List<PointOfTime> tsPoints = duration.assembleDurationPoints();
+ Map<Long, DataPoint> idMap = queryByEntityID(condition,
valueColumnName, duration, tsPoints.size());
HeatMap heatMap = new HeatMap();
if (idMap.isEmpty()) {
return heatMap;
}
- List<PointOfTime> tsPoints = duration.assembleDurationPoints();
String entityID = condition.getEntity().buildId();
List<String> ids = new ArrayList<>(tsPoints.size());
@@ -241,19 +240,24 @@ public class BanyanDBMetricsQueryDAO extends
AbstractBanyanDBDAO implements IMet
return heatMap;
}
- private Map<Long, DataPoint> queryByEntityID(final MetricsCondition
condition, String valueColumnName, Duration duration) throws IOException {
+ private Map<Long, DataPoint> queryByEntityID(final MetricsCondition
condition, String valueColumnName, Duration duration, int pointCount) throws
IOException {
final MetadataRegistry.Schema schema =
MetadataRegistry.INSTANCE.findMetricMetadata(condition.getName(),
duration.getStep());
if (schema == null) {
throw new IOException("schema is not registered");
}
- return queryByEntityID(schema, valueColumnName, duration,
condition.getEntity().buildId());
+ return queryByEntityID(schema, valueColumnName, duration,
condition.getEntity().buildId(), pointCount);
}
- private Map<Long, DataPoint> queryByEntityID(MetadataRegistry.Schema
schema, String valueColumnName, Duration duration, String entityID) throws
IOException {
+ private Map<Long, DataPoint> queryByEntityID(MetadataRegistry.Schema
schema, String valueColumnName, Duration duration, String entityID, int
pointCount) throws IOException {
final boolean isColdStage = duration != null && duration.isColdStage();
Map<Long, DataPoint> map = new HashMap<>();
+ // One entity over one time range yields at most one data point per
step, so the number of assembled
+ // duration points is the exact row cap — the same set of rows the
ES/JDBC DAOs fetch by explicit id.
+ // It must be sent: with no LIMIT the server falls back to its own
default of 100 and truncates any
+ // range longer than that (DurationUtils.MAX_TIME_RANGE allows up to
500 steps).
final Conditions where = Conditions.create()
- .eq(Metrics.ENTITY_ID, entityID);
+ .eq(Metrics.ENTITY_ID, entityID)
+ .limit(pointCount);
MeasureQueryResponse resp = queryDebuggable(isColdStage, schema,
ImmutableSet.of(Metrics.ENTITY_ID), ImmutableSet.of(valueColumnName),
getTimestampRange(duration),
where);
for (final DataPoint dp : resp.getDataPoints()) {
diff --git
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
index cffa0eb5ab..1a761a78f6 100644
---
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
+++
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/AbstractBanyanDBDAO.java
@@ -97,6 +97,7 @@ public abstract class AbstractBanyanDBDAO extends
AbstractDAO<BanyanDBStorageCli
ql.append(" ON
").append(BanyanDBStorageConfig.StageName.cold.name()).append(" STAGES");
}
ql.append(" TIME BETWEEN ? AND ?");
+ where.limitIfAbsent(getClient().getResultWindowMaxSize());
ql.append(where.buildQl(debug));
final List<Serializable<BanyandbModel.TagValue>> params =
timeBoundedParams(timestampRange, where.params());
final StreamQueryResponse response =
@@ -248,9 +249,14 @@ public abstract class AbstractBanyanDBDAO extends
AbstractDAO<BanyanDBStorageCli
if (debug) {
ql.append(" WITH QUERY_TRACE");
}
+ // TOP already bounds the output to `number` rows, but without an
explicit LIMIT the server caps the
+ // result at its own default (100) — which would silently shorten
a TopN request for more than that.
+ // The clause goes last: the grammar orders it after GROUP BY and
WITH QUERY_TRACE.
+ ql.append(" LIMIT ?");
final List<Serializable<BanyandbModel.TagValue>> params = new
ArrayList<>();
params.add(Value.longTagValue((long) number));
params.addAll(timeBoundedParams(timestampRange, where.params()));
+ params.add(Value.longTagValue((long) number));
final MeasureQueryResponse response =
getClient().queryMeasure(ql.toString(), params.toArray(new
Serializable[0]));
if (span != null) {
@@ -306,6 +312,7 @@ public abstract class AbstractBanyanDBDAO extends
AbstractDAO<BanyanDBStorageCli
ql.append(" ON
").append(BanyanDBStorageConfig.StageName.cold.name()).append(" STAGES");
}
ql.append(" TIME BETWEEN ? AND ?");
+ where.limitIfAbsent(getClient().getResultWindowMaxSize());
ql.append(where.buildQl(debug));
final List<Serializable<BanyandbModel.TagValue>> params =
timeBoundedParams(timestampRange, where.params());
final MeasureQueryResponse response =
@@ -418,6 +425,7 @@ public abstract class AbstractBanyanDBDAO extends
AbstractDAO<BanyanDBStorageCli
ql.append(" ON
").append(BanyanDBStorageConfig.StageName.cold.name()).append(" STAGES");
}
ql.append(" TIME BETWEEN ? AND ?");
+ where.limitIfAbsent(getClient().getResultWindowMaxSize());
ql.append(where.buildQl(debug));
final List<Serializable<BanyandbModel.TagValue>> params =
timeBoundedParams(timestampRange, where.params());
final TraceQueryResponse response =
diff --git
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/Conditions.java
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/Conditions.java
index fe7a8beaa1..14fe7fc8e3 100644
---
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/Conditions.java
+++
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/Conditions.java
@@ -42,6 +42,13 @@ public final class Conditions {
* position {@code WITH QUERY_TRACE} ahead of pagination without scanning
the generated text.
*/
private int paginationStart = -1;
+ /**
+ * Index in {@link #params} of the first pagination parameter, paired with
{@link #paginationStart} so
+ * {@link #limitIfAbsent(long)} can splice a {@code LIMIT} in ahead of an
existing {@code OFFSET} and keep
+ * the parameter order aligned with the placeholder order.
+ */
+ private int paginationParamIndex = -1;
+ private boolean limitSet;
private Conditions(boolean groupMode) {
this.groupMode = groupMode;
@@ -161,23 +168,58 @@ public final class Conditions {
}
public Conditions limit(long value) {
- if (paginationStart < 0) {
- paginationStart = ql.length();
- }
+ markPaginationStart();
ql.append(" LIMIT ?");
params.add(Value.longTagValue(value));
+ limitSet = true;
return this;
}
public Conditions offset(long value) {
- if (paginationStart < 0) {
- paginationStart = ql.length();
- }
+ markPaginationStart();
ql.append(" OFFSET ?");
params.add(Value.longTagValue(value));
return this;
}
+ /**
+ * Set {@code LIMIT} only if the caller has not already set one — the
backstop that keeps a query from
+ * inheriting BanyanDB's server-side default.
+ * <p>
+ * BanyanDB applies its own default when a request carries no limit: 100
rows for measures
+ * ({@code defaultLimit} in {@code
pkg/query/logical/measure/measure_analyzer.go}) and 20 for
+ * streams/traces. That default is applied <em>after</em> any {@code GROUP
BY}, so it silently truncates
+ * the result set rather than erroring — e.g. a 4-hour minute-step metrics
read returns only the first
+ * 100 points. OAP must therefore always send an explicit limit.
+ * <p>
+ * The clause is spliced in at the start of the pagination tail rather
than appended, so it still lands
+ * ahead of an {@code OFFSET} that was set first, as the grammar requires
({@code ... LIMIT ? OFFSET ?}).
+ *
+ * @param value the fallback row cap, used only when no {@code LIMIT} has
been set
+ * @return this builder
+ */
+ public Conditions limitIfAbsent(long value) {
+ if (limitSet) {
+ return this;
+ }
+ final boolean noPagination = paginationStart < 0;
+ final int qlPos = noPagination ? ql.length() : paginationStart;
+ final int paramPos = noPagination ? params.size() :
paginationParamIndex;
+ ql.insert(qlPos, " LIMIT ?");
+ params.add(paramPos, Value.longTagValue(value));
+ paginationStart = qlPos;
+ paginationParamIndex = paramPos;
+ limitSet = true;
+ return this;
+ }
+
+ private void markPaginationStart() {
+ if (paginationStart < 0) {
+ paginationStart = ql.length();
+ paginationParamIndex = params.size();
+ }
+ }
+
/**
* @return the assembled QL body ({@code WHERE ... ORDER BY ... LIMIT
...}), to be appended
* after the {@code SELECT ... FROM ...} projection.
diff --git
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/ConditionsTest.java
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/ConditionsTest.java
index f8ec24a82c..65a7033a21 100644
---
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/ConditionsTest.java
+++
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/ConditionsTest.java
@@ -196,6 +196,74 @@ public class ConditionsTest {
assertEquals(where.buildQl(), where.buildQl(false));
}
+ @Test
+ public void limitIfAbsentAppendsWhenNoLimitWasSet() {
+ final Conditions where = Conditions.create().eq("a",
"x").limitIfAbsent(500);
+ assertEquals(" WHERE a = ? LIMIT ?", where.buildQl());
+ final List<Serializable<BanyandbModel.TagValue>> params =
where.params();
+ assertEquals(2, params.size());
+ assertEquals("x", str(params.get(0)));
+ assertEquals(500L, lng(params.get(1)));
+ }
+
+ @Test
+ public void limitIfAbsentOnEmptyConditionsEmitsOnlyTheLimit() {
+ final Conditions where = Conditions.create().limitIfAbsent(7);
+ assertEquals(" LIMIT ?", where.buildQl());
+ assertEquals(7L, lng(where.params().get(0)));
+ }
+
+ @Test
+ public void limitIfAbsentKeepsTheCallerLimit() {
+ final Conditions where = Conditions.create().eq("a",
"x").limit(10).limitIfAbsent(500);
+ assertEquals(" WHERE a = ? LIMIT ?", where.buildQl());
+ final List<Serializable<BanyandbModel.TagValue>> params =
where.params();
+ assertEquals(2, params.size());
+ assertEquals(10L, lng(params.get(1)));
+ }
+
+ @Test
+ public void limitIfAbsentIsIdempotent() {
+ final Conditions where = Conditions.create().eq("a",
"x").limitIfAbsent(500).limitIfAbsent(900);
+ assertEquals(" WHERE a = ? LIMIT ?", where.buildQl());
+ assertEquals(500L, lng(where.params().get(1)));
+ }
+
+ @Test
+ public void limitIfAbsentSplicesAheadOfAnExistingOffset() {
+ // LIMIT must precede OFFSET in the grammar, so a fallback limit added
after an offset was set has to
+ // be inserted rather than appended — params move with it to stay
aligned with the placeholders.
+ final Conditions where = Conditions.create().eq("a",
"x").offset(20).limitIfAbsent(500);
+ assertEquals(" WHERE a = ? LIMIT ? OFFSET ?", where.buildQl());
+ final List<Serializable<BanyandbModel.TagValue>> params =
where.params();
+ assertEquals(3, params.size());
+ assertEquals("x", str(params.get(0)));
+ assertEquals(500L, lng(params.get(1)));
+ assertEquals(20L, lng(params.get(2)));
+ }
+
+ @Test
+ public void limitIfAbsentComposesAfterGroupByAndOrderBy() {
+ final Conditions where = Conditions.create()
+ .eq("a", "x")
+ .groupBy("g1")
+ .orderByDesc("t")
+ .limitIfAbsent(500);
+ assertEquals(" WHERE a = ? GROUP BY g1 ORDER BY t DESC LIMIT ?",
where.buildQl());
+ }
+
+ @Test
+ public void limitIfAbsentStillLeavesQueryTraceAheadOfPagination() {
+ final Conditions where = Conditions.create().eq("a",
"x").orderByDesc("t").limitIfAbsent(500);
+ assertEquals(" WHERE a = ? ORDER BY t DESC WITH QUERY_TRACE LIMIT ?",
where.buildQl(true));
+ }
+
+ @Test
+ public void limitIfAbsentSplicedBeforeOffsetStillLeavesQueryTraceAhead() {
+ final Conditions where = Conditions.create().eq("a",
"x").offset(20).limitIfAbsent(500);
+ assertEquals(" WHERE a = ? WITH QUERY_TRACE LIMIT ? OFFSET ?",
where.buildQl(true));
+ }
+
@Test
public void emptyInValuesAreRejectedLocally() {
assertThrows(IllegalArgumentException.class, () ->
Conditions.create().in("a", List.of()));
diff --git
a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/QueryLimitTest.java
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/QueryLimitTest.java
new file mode 100644
index 0000000000..55d859dd99
--- /dev/null
+++
b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/stream/QueryLimitTest.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.storage.plugin.banyandb.stream;
+
+import com.google.common.collect.ImmutableSet;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import org.apache.skywalking.library.banyandb.v1.client.MeasureQueryResponse;
+import org.apache.skywalking.library.banyandb.v1.client.TimestampRange;
+import org.apache.skywalking.library.banyandb.v1.client.metadata.Serializable;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
+import
org.apache.skywalking.oap.server.storage.plugin.banyandb.BanyanDBStorageClient;
+import
org.apache.skywalking.oap.server.storage.plugin.banyandb.MetadataRegistry;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * BanyanDB applies its own default LIMIT to any query that carries none — 100
rows for measures — and
+ * applies it after GROUP BY, so an over-long result set is silently truncated
rather than rejected. These
+ * tests pin the invariant that OAP never relies on that default: every query
leaves with an explicit LIMIT.
+ * The bound value and its placeholder ordering are covered by {@link
ConditionsTest}.
+ */
+public class QueryLimitTest {
+ private static final int RESULT_WINDOW_MAX_SIZE = 10000;
+
+ private final List<String> emitted = new ArrayList<>();
+ private BanyanDBStorageClient client;
+ private ProbeDAO dao;
+
+ /**
+ * Exposes the protected measure query helper — the single funnel every
measure DAO goes through.
+ */
+ private static class ProbeDAO extends AbstractBanyanDBDAO {
+ ProbeDAO(final BanyanDBStorageClient client) {
+ super(client);
+ }
+
+ void query(final MetadataRegistry.Schema schema, final Conditions
where) throws IOException {
+ queryDebuggable(false, schema, ImmutableSet.of("entity_id"),
Collections.emptySet(),
+ new TimestampRange(0, 1), where);
+ }
+ }
+
+ @BeforeEach
+ public void setUp() throws IOException {
+ client = mock(BanyanDBStorageClient.class);
+
when(client.getResultWindowMaxSize()).thenReturn(RESULT_WINDOW_MAX_SIZE);
+ // queryMeasure is varargs — the matcher has to target the array type,
not a single element.
+ when(client.queryMeasure(anyString(),
any(Serializable[].class))).thenAnswer(invocation -> {
+ emitted.add(invocation.getArgument(0));
+ return mock(MeasureQueryResponse.class);
+ });
+ dao = new ProbeDAO(client);
+ }
+
+ private static MetadataRegistry.Schema schema() {
+ return MetadataRegistry.Schema.builder()
+ .metadata(new
MetadataRegistry.SchemaMetadata(
+ "sw", "measure-default",
"service_cpm",
+ MetadataRegistry.Kind.MEASURE,
DownSampling.Minute, null))
+ .build();
+ }
+
+ @Test
+ public void measureQueryWithoutAnExplicitLimitFallsBackToTheResultWindow()
throws IOException {
+ dao.query(schema(), Conditions.create().eq("entity_id", "svc"));
+
+ assertEquals(1, emitted.size());
+ assertTrue(emitted.get(0).endsWith(" LIMIT ?"),
+ "a query with no caller limit must still carry one, was: "
+ emitted.get(0));
+ verify(client, times(1)).getResultWindowMaxSize();
+ }
+
+ @Test
+ public void measureQueryWithNoConditionsAtAllStillCarriesALimit() throws
IOException {
+ // The whole-range topology reads pass an empty condition set; they
were the worst hit by the
+ // server-side default because they return one row per relation.
+ dao.query(schema(), Conditions.create());
+
+ assertEquals(1, emitted.size());
+ assertTrue(emitted.get(0).endsWith(" LIMIT ?"),
+ "an unconditional query must still carry a limit, was: " +
emitted.get(0));
+ }
+
+ @Test
+ public void measureQueryKeepsAnExplicitLimit() throws IOException {
+ dao.query(schema(), Conditions.create().eq("entity_id",
"svc").limit(240));
+
+ assertEquals(1, countOccurrences(emitted.get(0), "LIMIT"),
+ "the caller's limit must not be doubled, was: " +
emitted.get(0));
+ }
+
+ @Test
+ public void groupedQueryPutsTheFallbackLimitAfterGroupBy() throws
IOException {
+ dao.query(schema(), Conditions.create().eq("entity_id",
"svc").groupBy("entity_id"));
+
+ assertTrue(emitted.get(0).endsWith(" GROUP BY entity_id LIMIT ?"),
+ "LIMIT must follow GROUP BY, was: " + emitted.get(0));
+ }
+
+ private static int countOccurrences(final String text, final String token)
{
+ int count = 0;
+ int idx = text.indexOf(token);
+ while (idx >= 0) {
+ count++;
+ idx = text.indexOf(token, idx + token.length());
+ }
+ return count;
+ }
+}
