This is an automated email from the ASF dual-hosted git repository. wu-sheng pushed a commit to branch fix/banyandb-implicit-query-limit-truncation in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 8bf7041635af72ea43f2d9909a9a9ebc41c05098 Author: Wu Sheng <[email protected]> AuthorDate: Wed Jul 29 06:33:38 2026 +0800 Always send an explicit LIMIT on BanyanDB queries 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; + } +}
