You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wa...@apache.org on 2026/05/21 06:23:56 UTC
(skywalking) branch master updated: Add `@Stream(allowBootReshape = true)` opt-in for additive boot-time reshape of BanyanDB streams / measures. (#13878)
This is an automated email from the ASF dual-hosted git repository.
wankai123 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 5e9db8fba4 Add `@Stream(allowBootReshape = true)` opt-in for additive boot-time reshape of BanyanDB streams / measures. (#13878)
5e9db8fba4 is described below
commit 5e9db8fba4c6b11620137afcbf9ccbd788c1214d
Author: Wan Kai <wa...@foxmail.com>
AuthorDate: Thu May 21 14:23:51 2026 +0800
Add `@Stream(allowBootReshape = true)` opt-in for additive boot-time reshape of BanyanDB streams / measures. (#13878)
---
CLAUDE.md | 4 +
NOTICE | 2 +-
dist-material/release-docs/NOTICE | 2 +-
docs/en/changes/changes.md | 1 +
.../runtime-rule-hot-update.md | 21 +
.../oap/server/core/alarm/AlarmRecord.java | 2 +-
.../oap/server/core/analysis/Stream.java | 26 +
.../oap/server/core/storage/model/Model.java | 10 +
.../server/core/storage/model/StorageModels.java | 5 +
.../plugin/banyandb/BanyanDBIndexInstaller.java | 133 +++-
.../server/storage/plugin/banyandb/BanyanDBIT.java | 882 +++++++++++++++++++++
.../src/test/resources/bydb.yml | 4 +
.../elasticsearch/base/TimeSeriesUtilsTest.java | 6 +-
.../jdbc/common/dao/JDBCHistoryDeleteDAOIT.java | 2 +-
14 files changed, 1087 insertions(+), 13 deletions(-)
diff --git a/CLAUDE.md b/CLAUDE.md
index dacd9a41ba..1608fda657 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -246,6 +246,10 @@ Always use `--recurse-submodules` when cloning or update submodules manually.
- `changes/changes.md` - Changelog (update when making changes)
- `swip/` - SkyWalking Improvement Proposals
+## Handling Existing Tests
+
+If existing test code (UT, IT, or E2E) fails to pass, try to fix it. If any test cases need to be modified, or test code needs to be deleted, the change must be double-checked by a human before it is applied — surface the proposed modification or deletion and the reasoning, and wait for explicit approval rather than changing or removing the test yourself.
+
## Submitting Pull Requests
Use the `/gh-pull-request` skill for committing and pushing to a PR branch. It runs pre-flight checks (compile, checkstyle, license headers) before every push, and creates the PR if one doesn't exist yet.
diff --git a/NOTICE b/NOTICE
index 7e684bf85d..0fef61473b 100644
--- a/NOTICE
+++ b/NOTICE
@@ -1,5 +1,5 @@
Apache SkyWalking
-Copyright 2017-2024 The Apache Software Foundation
+Copyright 2017-2026 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
diff --git a/dist-material/release-docs/NOTICE b/dist-material/release-docs/NOTICE
index 6c5c72afe1..90fdf3a8b5 100755
--- a/dist-material/release-docs/NOTICE
+++ b/dist-material/release-docs/NOTICE
@@ -1,5 +1,5 @@
Apache SkyWalking
-Copyright 2017-2024 The Apache Software Foundation
+Copyright 2017-2026 The Apache Software Foundation
This product includes software developed at
The Apache Software Foundation (http://www.apache.org/).
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index ddf89306e5..322ada7e4d 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -279,6 +279,7 @@
* LAL: support full arithmetic (`+`, `-`, `*`, `/`) on numeric operands and fix the original bug where `(tag("x") as Integer) + (tag("y") as Integer)` was treated as string concatenation — expressions like `input_tokens + output_tokens < 10000` produced the concatenated string `"2589115"` rather than the integer sum `2704`, so token-threshold conditions never triggered `abort {}`. Operand types are now inferred from explicit casts (`as Integer` / `as Long` / `as Float` / `as Double`), ty [...]
* Fix: `avgHistogramPercentile` / `sumHistogramPercentile` meter functions reported the smallest finite bucket boundary (e.g. `10` for OTel `gen_ai_server_request_duration` whose `le` is rewritten from `0.01s` → `10ms`) for every rank when no samples were observed in any bucket. The percentile loop's `count >= roof` check matched on the first sorted bucket because both sides were `0`. `calculate()` now short-circuits to `0` for every rank when the windowed total is `0`.
* Fix: MAL `expPrefix` now applies to every metric source in `exp`, not just the leading one. Previously the prefix was spliced after the first `.`, so secondary metrics inside arguments (e.g. the divisor in `a.sum(['s']).safeDiv(b.sum(['s']))`) silently skipped the prefix — a rule like envoy-ai-gateway's `request_latency_avg` (`sum / count`) would tag-rewrite only the dividend. The injection is now AST-aware: every bare-IDENTIFIER metric source is wrapped, while downsampling-type consta [...]
+* Add `@Stream(allowBootReshape = true)` opt-in for additive boot-time reshape of BanyanDB streams / measures. Code-defined stream classes (e.g. `AlarmRecord`) can now annotate their schema as eligible for in-place additive update at OAP boot — a new `@Column` is appended to the live tag-family / fields via `client.update` instead of being silently rejected with `SKIPPED_SHAPE_MISMATCH` (which previously forced operators to drop the measure / stream and lose historical rows). The opt-in [...]
#### UI
* Add mobile menu icon and i18n labels for the iOS layer.
diff --git a/docs/en/concepts-and-designs/runtime-rule-hot-update.md b/docs/en/concepts-and-designs/runtime-rule-hot-update.md
index f1901cc6e9..d20fc5ed86 100644
--- a/docs/en/concepts-and-designs/runtime-rule-hot-update.md
+++ b/docs/en/concepts-and-designs/runtime-rule-hot-update.md
@@ -160,6 +160,27 @@ explicit ERROR, boot always continues, the affected metric is disabled, and the
operator reconciles explicitly through the on-demand workflow. The same shape is
visible across every backend.
+### Code-defined stream opt-in (narrow exception)
+
+The "boot never reshapes" rule above applies to **runtime-rule (MAL / LAL)**
+registration — those rules ride the `/addOrUpdate` REST workflow when their
+backing schema needs to change.
+
+Streams whose schema lives in OAP source code (e.g. `AlarmRecord`) can opt in
+to **additive** boot-time reshape via
+`@Stream(allowBootReshape = true)`. When the flag is on and the diff is
+purely additive (new tag / new field; no type changes, no drops, no entity /
+interval / sharding-key flips), the installer calls `client.update` at boot
+to append the new column to the live measure / stream; non-additive
+divergences still record `SKIPPED_SHAPE_MISMATCH` and require an operator
+drop+recreate. Only the init / standalone OAP performs the reshape; non-init
+peers continue through the existing poll-and-wait loop so a single node
+drives DDL during a rolling restart.
+
+This opt-in is **BanyanDB-only**. JDBC and Elasticsearch are append-only on
+the data path and already accept additive column / mapping additions at boot
+without operator intervention, so the flag is unread on those backends.
+
## On-demand workflow
Triggered by an HTTP call to one of the admin endpoints. A request arriving at any
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmRecord.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmRecord.java
index eafde843a7..a5a1e474b3 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmRecord.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/alarm/AlarmRecord.java
@@ -42,7 +42,7 @@ import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.AL
@Getter
@Setter
@ScopeDeclaration(id = ALARM, name = "Alarm")
-@Stream(name = AlarmRecord.INDEX_NAME, scopeId = DefaultScopeDefine.ALARM, builder = AlarmRecord.Builder.class, processor = RecordStreamProcessor.class)
+@Stream(name = AlarmRecord.INDEX_NAME, scopeId = DefaultScopeDefine.ALARM, builder = AlarmRecord.Builder.class, processor = RecordStreamProcessor.class, allowBootReshape = true)
@SQLDatabase.ExtraColumn4AdditionalEntity(additionalTable = AlarmRecord.ADDITIONAL_TAG_TABLE, parentColumn = TIME_BUCKET)
@BanyanDB.TimestampColumn(AlarmRecord.START_TIME)
@BanyanDB.Group(streamGroup = BanyanDB.StreamGroup.RECORDS)
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/Stream.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/Stream.java
index acab363060..ab00568eb0 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/Stream.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/Stream.java
@@ -27,6 +27,8 @@ import org.apache.skywalking.oap.server.core.analysis.worker.NoneStreamProcessor
import org.apache.skywalking.oap.server.core.analysis.worker.RecordStreamProcessor;
import org.apache.skywalking.oap.server.core.analysis.worker.TopNStreamProcessor;
import org.apache.skywalking.oap.server.core.source.ScopeDeclaration;
+import org.apache.skywalking.oap.server.core.storage.model.ModelInstaller;
+import org.apache.skywalking.oap.server.core.storage.model.StorageManipulationOpt;
import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
/**
@@ -59,4 +61,28 @@ public @interface Stream {
* TopNStreamProcessor} and {@link NoneStreamProcessor} for more details.
*/
Class<? extends StreamProcessor> processor();
+
+ /**
+ * Opt-in to additive boot-time reshape of the backend resource for this stream class.
+ * <strong>BanyanDB only</strong> — JDBC and Elasticsearch are append-only on the data
+ * path and don't refuse additive column / mapping additions at boot, so they don't
+ * need (and don't read) this flag.
+ *
+ * <p>Default is {@code false} — boot follows the standard {@code schemaCreateIfAbsent}
+ * policy and records {@link StorageManipulationOpt.Outcome#SKIPPED_SHAPE_MISMATCH} on
+ * any shape divergence, leaving reconciliation to the operator (the only workflow that
+ * may change backend schema).
+ *
+ * <p>When set to {@code true}, the BanyanDB installer is allowed to apply
+ * <strong>purely additive</strong> changes (new tag, new field) during boot via
+ * {@code client.update}. Type changes, drops, kind flips, entity / interval /
+ * sharding-key changes are still refused with {@code SKIPPED_SHAPE_MISMATCH} and
+ * require a manual drop+recreate (identity-breaking edits are explicit operator
+ * actions, not boot side-effects).
+ *
+ * <p>Only the init / standalone OAP performs the reshape; non-init peers continue
+ * through the existing poll-and-wait loop in {@link ModelInstaller#whenCreating} so a
+ * single node drives DDL during a rolling restart.
+ */
+ boolean allowBootReshape() default false;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/Model.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/Model.java
index 5ceb6294d2..ed139cc418 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/Model.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/Model.java
@@ -41,6 +41,16 @@ public class Model {
private final boolean superDataset;
private final Class<?> streamClass;
private final boolean timeRelativeID;
+ /**
+ * BanyanDB only — when true, the BanyanDB installer is allowed to apply purely
+ * additive shape changes (new tag / new field) at boot. See
+ * {@link org.apache.skywalking.oap.server.core.analysis.Stream#allowBootReshape()}.
+ * JDBC and Elasticsearch ignore this flag (append-only data paths already accept
+ * additive column / mapping additions without operator intervention). Defaults to
+ * false for models registered without a {@code @Stream} annotation (e.g. runtime-rule
+ * MAL / LAL metrics, which reshape through the runtime-rule REST path instead).
+ */
+ private final boolean allowBootReshape;
private final SQLDatabaseModelExtension sqlDBModelExtension;
private final BanyanDBModelExtension banyanDBModelExtension;
private final ElasticSearchModelExtension elasticSearchModelExtension;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
index 558b47378f..72e8ce57e3 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/model/StorageModels.java
@@ -18,6 +18,7 @@
package org.apache.skywalking.oap.server.core.storage.model;
import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.analysis.Stream;
import org.apache.skywalking.oap.server.core.analysis.record.Record;
import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
import org.apache.skywalking.oap.server.core.storage.StorageException;
@@ -169,6 +170,9 @@ public class StorageModels implements IModelManager, ModelRegistry, ModelManipul
seriesIDChecker.check(storage.getModelName());
shardingKeyChecker.check(storage.getModelName());
+ final Stream streamAnnotation = aClass.getAnnotation(Stream.class);
+ final boolean allowBootReshape = streamAnnotation != null && streamAnnotation.allowBootReshape();
+
Model model = new Model(
storage.getModelName(),
modelColumns,
@@ -177,6 +181,7 @@ public class StorageModels implements IModelManager, ModelRegistry, ModelManipul
isSuperDatasetModel(aClass),
aClass,
storage.isTimeRelativeID(),
+ allowBootReshape,
sqlDBModelExtension,
banyanDBModelExtension,
elasticSearchModelExtension
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java
index abf4f288e3..092214919e 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIndexInstaller.java
@@ -172,7 +172,7 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
StreamModel streamModel = MetadataRegistry.INSTANCE.registerStreamModel(
model, config);
if (runShapeChecks) {
- checkStream(streamModel.getStream(), c, opt);
+ checkStream(model, streamModel.getStream(), c, opt);
checkIndexRules(model.getName(), streamModel.getIndexRules(), c, opt);
checkIndexRuleBinding(
streamModel.getIndexRules(), metadata.getGroup(), metadata.name(),
@@ -184,7 +184,7 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
} else { // measure
MeasureModel measureModel = MetadataRegistry.INSTANCE.registerMeasureModel(model, config, downSamplingConfigService);
if (runShapeChecks) {
- checkMeasure(measureModel.getMeasure(), c, opt);
+ checkMeasure(model, measureModel.getMeasure(), c, opt);
checkIndexRules(model.getName(), measureModel.getIndexRules(), c, opt);
checkIndexRuleBinding(
measureModel.getIndexRules(), metadata.getGroup(), metadata.name(),
@@ -767,9 +767,16 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
* either update it (on-demand operator workflow — {@link StorageManipulationOpt#isWithSchemaChange()})
* or skip the update and record {@link StorageManipulationOpt.Outcome#SKIPPED_SHAPE_MISMATCH}
* (static boot workflow — {@link StorageManipulationOpt#isSchemaCreateIfAbsent()}). Boot MUST
- * NOT reshape the backend — reshape is an explicit operator action only.
+ * NOT reshape the backend by default — reshape is an explicit operator action.
+ *
+ * <p>Exception: when the model opts in via {@link Model#isAllowBootReshape()} and the diff
+ * is purely additive (new tag / new field, no type changes, no drops, identity preserved),
+ * the init OAP is allowed to apply the additive update during boot. Non-init OAPs continue
+ * through the poll-and-wait loop in
+ * {@link org.apache.skywalking.oap.server.core.storage.model.ModelInstaller#whenCreating}
+ * so only one node races on the DDL.
*/
- private void checkMeasure(Measure measure, BanyanDBClient client, StorageManipulationOpt opt) throws BanyanDBException {
+ private void checkMeasure(Model model, Measure measure, BanyanDBClient client, StorageManipulationOpt opt) throws BanyanDBException {
Measure hisMeasure = client.findMeasure(measure.getMetadata().getGroup(), measure.getMetadata().getName());
if (hisMeasure == null) {
throw new IllegalStateException("Measure: " + measure.getMetadata().getName() + " exist but can't find it from BanyanDB server");
@@ -782,6 +789,15 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
.equals(measure.toBuilder().clearMetadata().build());
if (!equals) {
if (!opt.getFlags().isUpdateOnMismatch()) {
+ if (canBootReshape(model, opt) && isPurelyAdditiveMeasure(measure, hisMeasure)) {
+ opt.recordModRevision(client.update(measure));
+ log.info("boot reshape (additive) Measure: {} — applied @Stream(allowBootReshape=true). backend={}, declared={}",
+ hisMeasure.getMetadata().getName(), hisMeasure, measure);
+ opt.recordOutcome("measure", hisMeasure.getMetadata().getName(),
+ StorageManipulationOpt.Outcome.UPDATED,
+ "additive boot reshape: new tag / field added");
+ return;
+ }
log.error("BanyanDB measure {} shape mismatch at boot — backend holds a "
+ "different shape than the declared rule. SKIPPING metric; operator "
+ "must reshape via POST /runtime/rule/addOrUpdate or align the rule "
@@ -801,9 +817,10 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
/**
* Check if the stream exists and update (or record shape mismatch) per mode.
- * See {@link #checkMeasure} for the create-if-absent vs full-install contract.
+ * See {@link #checkMeasure} for the create-if-absent vs full-install contract,
+ * including the {@link Model#isAllowBootReshape()} additive opt-in.
*/
- private void checkStream(Stream stream, BanyanDBClient client, StorageManipulationOpt opt) throws BanyanDBException {
+ private void checkStream(Model model, Stream stream, BanyanDBClient client, StorageManipulationOpt opt) throws BanyanDBException {
Stream hisStream = client.findStream(stream.getMetadata().getGroup(), stream.getMetadata().getName());
if (hisStream == null) {
throw new IllegalStateException("Stream: " + stream.getMetadata().getName() + " exist but can't find it from BanyanDB server");
@@ -816,6 +833,15 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
.equals(stream.toBuilder().clearUpdatedAt().clearCreatedAt().clearMetadata().build());
if (!equals) {
if (!opt.getFlags().isUpdateOnMismatch()) {
+ if (canBootReshape(model, opt) && isPurelyAdditiveStream(stream, hisStream)) {
+ opt.recordModRevision(client.update(stream));
+ log.info("boot reshape (additive) Stream: {} — applied @Stream(allowBootReshape=true). backend={}, declared={}",
+ hisStream.getMetadata().getName(), hisStream, stream);
+ opt.recordOutcome("stream", hisStream.getMetadata().getName(),
+ StorageManipulationOpt.Outcome.UPDATED,
+ "additive boot reshape: new tag added");
+ return;
+ }
log.error("BanyanDB stream {} shape mismatch at boot — backend holds a "
+ "different shape than the declared rule. SKIPPING; operator must "
+ "reshape via POST /runtime/rule/addOrUpdate. backend={}, declared={}",
@@ -831,6 +857,101 @@ public class BanyanDBIndexInstaller extends ModelInstaller {
}
}
+ /**
+ * Gate for boot-time reshape: three conditions, all required.
+ * <ul>
+ * <li>The model opts in via {@link Model#isAllowBootReshape()}.</li>
+ * <li>The opt is the static-boot policy ({@link StorageManipulationOpt#isSchemaCreateIfAbsent()}).
+ * Restricting the reshape branch to this single mode keeps the policy boundary
+ * explicit — {@code verifySchemaOnly()} must stay read-only even if a future
+ * caller flips {@code updateOnMismatch} off, and {@code withSchemaChange()}
+ * already takes the existing on-demand reshape path above.</li>
+ * <li>This OAP must not be in {@code no-init} mode. Non-init OAPs leave DDL to the
+ * init / standalone OAP and converge via the poll-wait loop in
+ * {@link org.apache.skywalking.oap.server.core.storage.model.ModelInstaller#whenCreating}.</li>
+ * </ul>
+ */
+ private boolean canBootReshape(Model model, StorageManipulationOpt opt) {
+ return model.isAllowBootReshape()
+ && opt.isSchemaCreateIfAbsent()
+ && !RunningMode.isNoInitMode();
+ }
+
+ /**
+ * Purely-additive diff for a BanyanDB {@link Stream}: declared may add tag families or
+ * tags, but every tag-family / tag that already exists on the backend must be present
+ * with the same name and {@link BanyandbDatabase.TagType type}, the {@link BanyandbDatabase.Entity entity}
+ * column list must match exactly (reshape can't change shard / series-id semantics),
+ * and no tag may be dropped. Returns false for any non-additive divergence so the caller
+ * falls back to {@link StorageManipulationOpt.Outcome#SKIPPED_SHAPE_MISMATCH}.
+ */
+ private boolean isPurelyAdditiveStream(Stream declared, Stream live) {
+ if (!declared.getEntity().equals(live.getEntity())) {
+ return false;
+ }
+ return isPurelyAdditiveTagFamilies(declared.getTagFamiliesList(), live.getTagFamiliesList());
+ }
+
+ /**
+ * Purely-additive diff for a BanyanDB {@link Measure}: same tag-family and entity rules as
+ * {@link #isPurelyAdditiveStream}, plus fields may be added but no field name / type /
+ * encoding / compression may change, and the scalar properties (interval, index_mode,
+ * sharding_key) must match exactly.
+ */
+ private boolean isPurelyAdditiveMeasure(Measure declared, Measure live) {
+ if (!declared.getEntity().equals(live.getEntity())) {
+ return false;
+ }
+ if (!declared.getInterval().equals(live.getInterval())) {
+ return false;
+ }
+ if (declared.getIndexMode() != live.getIndexMode()) {
+ return false;
+ }
+ if (!declared.getShardingKey().equals(live.getShardingKey())) {
+ return false;
+ }
+ if (!isPurelyAdditiveTagFamilies(declared.getTagFamiliesList(), live.getTagFamiliesList())) {
+ return false;
+ }
+ final Map<String, BanyandbDatabase.FieldSpec> declaredFields = declared.getFieldsList().stream()
+ .collect(Collectors.toMap(BanyandbDatabase.FieldSpec::getName, f -> f, (a, b) -> a));
+ for (BanyandbDatabase.FieldSpec liveField : live.getFieldsList()) {
+ BanyandbDatabase.FieldSpec declaredField = declaredFields.get(liveField.getName());
+ if (declaredField == null) {
+ return false;
+ }
+ if (!declaredField.equals(liveField)) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private boolean isPurelyAdditiveTagFamilies(List<BanyandbDatabase.TagFamilySpec> declared,
+ List<BanyandbDatabase.TagFamilySpec> live) {
+ final Map<String, BanyandbDatabase.TagFamilySpec> declaredByName = declared.stream()
+ .collect(Collectors.toMap(BanyandbDatabase.TagFamilySpec::getName, f -> f, (a, b) -> a));
+ for (BanyandbDatabase.TagFamilySpec liveFamily : live) {
+ BanyandbDatabase.TagFamilySpec declaredFamily = declaredByName.get(liveFamily.getName());
+ if (declaredFamily == null) {
+ return false;
+ }
+ final Map<String, BanyandbDatabase.TagSpec> declaredTags = declaredFamily.getTagsList().stream()
+ .collect(Collectors.toMap(BanyandbDatabase.TagSpec::getName, t -> t, (a, b) -> a));
+ for (BanyandbDatabase.TagSpec liveTag : liveFamily.getTagsList()) {
+ BanyandbDatabase.TagSpec declaredTag = declaredTags.get(liveTag.getName());
+ if (declaredTag == null) {
+ return false;
+ }
+ if (declaredTag.getType() != liveTag.getType()) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
private void checkTrace(Trace trace, BanyanDBClient client, StorageManipulationOpt opt) throws BanyanDBException {
Trace hisTrace = client.findTrace(trace.getMetadata().getGroup(), trace.getMetadata().getName());
if (hisTrace == null) {
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIT.java b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIT.java
new file mode 100644
index 0000000000..58e19d4f0a
--- /dev/null
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/banyandb/BanyanDBIT.java
@@ -0,0 +1,882 @@
+/*
+ * 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;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.banyandb.common.v1.BanyandbCommon;
+import org.apache.skywalking.banyandb.database.v1.BanyandbDatabase;
+import org.apache.skywalking.banyandb.model.v1.BanyandbModel;
+import org.apache.skywalking.library.banyandb.v1.client.DataPoint;
+import org.apache.skywalking.library.banyandb.v1.client.MeasureQuery;
+import org.apache.skywalking.library.banyandb.v1.client.MeasureQueryResponse;
+import org.apache.skywalking.library.banyandb.v1.client.MeasureWrite;
+import org.apache.skywalking.library.banyandb.v1.client.TagAndValue;
+import org.apache.skywalking.library.banyandb.v1.client.TimestampRange;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.analysis.DownSampling;
+import org.apache.skywalking.oap.server.core.analysis.Stream;
+import org.apache.skywalking.oap.server.core.analysis.record.Record;
+import org.apache.skywalking.oap.server.core.analysis.worker.MetricsStreamProcessor;
+import org.apache.skywalking.oap.server.core.analysis.worker.RecordStreamProcessor;
+import org.apache.skywalking.oap.server.core.config.DownSamplingConfigService;
+import org.apache.skywalking.oap.server.core.source.DefaultScopeDefine;
+import org.apache.skywalking.oap.server.core.storage.StorageData;
+import org.apache.skywalking.oap.server.core.storage.StorageID;
+import org.apache.skywalking.oap.server.core.storage.annotation.BanyanDB;
+import org.apache.skywalking.oap.server.core.storage.annotation.Column;
+import org.apache.skywalking.oap.server.core.storage.annotation.Storage;
+import org.apache.skywalking.oap.server.core.storage.model.Model;
+import org.apache.skywalking.oap.server.core.storage.model.StorageManipulationOpt;
+import org.apache.skywalking.oap.server.core.storage.model.StorageModels;
+import org.apache.skywalking.oap.server.core.storage.type.Convert2Entity;
+import org.apache.skywalking.oap.server.core.storage.type.Convert2Storage;
+import org.apache.skywalking.oap.server.core.storage.type.StorageBuilder;
+import org.apache.skywalking.oap.server.library.it.ITVersions;
+import org.apache.skywalking.oap.server.library.module.ModuleDefine;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.module.ModuleProviderHolder;
+import org.apache.skywalking.oap.server.library.module.ModuleServiceHolder;
+import org.apache.skywalking.oap.server.storage.plugin.banyandb.bulk.MeasureBulkWriteProcessor;
+import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
+import org.apache.skywalking.oap.server.telemetry.none.MetricsCreatorNoop;
+import org.apache.skywalking.oap.server.telemetry.none.NoneTelemetryProvider;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import org.apache.skywalking.oap.server.testing.util.ReflectUtil;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+import static org.testcontainers.shaded.org.awaitility.Awaitility.await;
+
+@Slf4j
+@Testcontainers
+public class BanyanDBIT {
+ private static final String REGISTRY = "ghcr.io";
+ private static final String IMAGE_NAME = "apache/skywalking-banyandb";
+ private static final String TAG = ITVersions.get("SW_BANYANDB_COMMIT");
+
+ private static final String IMAGE = REGISTRY + "/" + IMAGE_NAME + ":" + TAG;
+ private static MockedStatic<DefaultScopeDefine> DEFAULT_SCOPE_DEFINE_MOCKED_STATIC;
+ protected static final int GRPC_PORT = 17912;
+ protected static final int HTTP_PORT = 17913;
+
+ @Container
+ public GenericContainer<?> banyanDB = new GenericContainer<>(
+ DockerImageName.parse(IMAGE))
+ .withCommand("standalone", "--stream-root-path", "/tmp/banyandb-stream-data",
+ "--measure-root-path", "/tmp/banyand-measure-data"
+ )
+ .withExposedPorts(GRPC_PORT, HTTP_PORT)
+ .waitingFor(Wait.forHttp("/api/healthz").forPort(HTTP_PORT));
+
+ private BanyanDBStorageClient client;
+ private BanyanDBStorageConfig config;
+
+ protected void setUpConnection() throws Exception {
+ ModuleManager moduleManager = mock(ModuleManager.class);
+ ModuleDefine storageModule = mock(ModuleDefine.class);
+ BanyanDBStorageProvider provider = mock(BanyanDBStorageProvider.class);
+ Mockito.when(provider.getModule()).thenReturn(storageModule);
+
+ NoneTelemetryProvider telemetryProvider = mock(NoneTelemetryProvider.class);
+ Mockito.when(telemetryProvider.getService(MetricsCreator.class))
+ .thenReturn(new MetricsCreatorNoop());
+ TelemetryModule telemetryModule = Mockito.spy(TelemetryModule.class);
+ ReflectUtil.setInternalState(telemetryModule, "loadedProvider", telemetryProvider);
+ Mockito.when(moduleManager.find(TelemetryModule.NAME)).thenReturn(telemetryModule);
+ log.info("create BanyanDB client and try to connect");
+ config = new BanyanDBConfigLoader(provider).loadConfig();
+ config.getGlobal().setTargets(banyanDB.getHost() + ":" + banyanDB.getMappedPort(GRPC_PORT));
+ client = new BanyanDBStorageClient(moduleManager, config);
+ client.connect();
+ }
+
+ private MeasureBulkWriteProcessor processor;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ DEFAULT_SCOPE_DEFINE_MOCKED_STATIC = mockStatic(DefaultScopeDefine.class);
+ DEFAULT_SCOPE_DEFINE_MOCKED_STATIC.when(() -> DefaultScopeDefine.nameOf(1)).thenReturn("any");
+ setUpConnection();
+ processor = client.createMeasureBulkProcessor(1000, 1, 1);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (DEFAULT_SCOPE_DEFINE_MOCKED_STATIC != null) {
+ DEFAULT_SCOPE_DEFINE_MOCKED_STATIC.close();
+ DEFAULT_SCOPE_DEFINE_MOCKED_STATIC = null;
+ }
+ }
+
+ @Test
+ public void testInstall() throws Exception {
+ DownSamplingConfigService downSamplingConfigService = new DownSamplingConfigService(Arrays.asList("minute"));
+ ModuleManager moduleManager = mock(ModuleManager.class);
+ ModuleProviderHolder moduleProviderHolder = mock(ModuleProviderHolder.class);
+ ModuleServiceHolder moduleServiceHolder = mock(ModuleServiceHolder.class);
+ when(moduleManager.find(CoreModule.NAME)).thenReturn(moduleProviderHolder);
+ when(moduleProviderHolder.provider()).thenReturn(moduleServiceHolder);
+ when(moduleServiceHolder.getService(DownSamplingConfigService.class)).thenReturn(downSamplingConfigService);
+
+ StorageModels models = new StorageModels();
+ Model model = models.add(TestMetric.class, DefaultScopeDefine.SERVICE,
+ new Storage("testMetric", true, DownSampling.Minute),
+ StorageManipulationOpt.withSchemaChange()
+ );
+ BanyanDBIndexInstaller installer = new BanyanDBIndexInstaller(client, moduleManager, config);
+ installer.isExists(model, StorageManipulationOpt.withSchemaChange());
+ //test Group install
+ String groupName = MetadataRegistry.convertGroupName(
+ config.getGlobal().getNamespace(),
+ BanyanDB.MeasureGroup.METRICS_MINUTE.getName()
+ );
+ BanyandbCommon.Group group = client.client.findGroup(groupName);
+ assertEquals(BanyandbCommon.Catalog.CATALOG_MEASURE, group.getCatalog());
+ assertEquals(config.getMetricsMin().getSegmentInterval(), group.getResourceOpts().getSegmentInterval().getNum());
+ assertEquals(config.getMetricsMin().getShardNum(), group.getResourceOpts().getShardNum());
+ assertEquals(BanyandbCommon.IntervalRule.Unit.UNIT_DAY, group.getResourceOpts().getSegmentInterval().getUnit());
+ assertEquals(config.getMetricsMin().getTtl(), group.getResourceOpts().getTtl().getNum());
+ assertEquals(BanyandbCommon.IntervalRule.Unit.UNIT_DAY, group.getResourceOpts().getTtl().getUnit());
+
+ installer.createTable(model);
+ //test Measure install
+ BanyandbDatabase.Measure measure = client.client.findMeasure(groupName, "testMetric_minute");
+ assertEquals("storage-only", measure.getTagFamilies(0).getName());
+ assertEquals("service_id", measure.getTagFamilies(0).getTags(0).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, measure.getTagFamilies(0).getTags(0).getType());
+ assertEquals("searchable", measure.getTagFamilies(1).getName());
+ assertEquals("tag", measure.getTagFamilies(1).getTags(0).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, measure.getTagFamilies(1).getTags(0).getType());
+ assertEquals("service_id", measure.getEntity().getTagNames(0));
+ assertEquals("value", measure.getFields(0).getName());
+ assertEquals(BanyandbDatabase.FieldType.FIELD_TYPE_INT, measure.getFields(0).getFieldType());
+ //test TopNAggregation install
+ BanyandbDatabase.TopNAggregation topNAggregation = client.client.findTopNAggregation(
+ groupName, "testMetric-service");
+ assertEquals("value", topNAggregation.getFieldName());
+ assertEquals("service_id", topNAggregation.getGroupByTagNames(0));
+ assertEquals(BanyandbModel.Sort.SORT_DESC, topNAggregation.getFieldValueSort());
+ assertEquals(10, topNAggregation.getLruSize());
+ assertEquals(1000, topNAggregation.getCountersNumber());
+ //test IndexRule install
+ BanyandbDatabase.IndexRule indexRuleTag = client.client.findIndexRule(groupName, "tag");
+ assertEquals("url", indexRuleTag.getAnalyzer());
+ assertTrue(indexRuleTag.getNoSort());
+ //test IndexRuleBinding install
+ BanyandbDatabase.IndexRuleBinding indexRuleBinding = client.client.findIndexRuleBinding(
+ groupName, "testMetric_minute");
+ assertEquals("tag", indexRuleBinding.getRules(0));
+ assertEquals("testMetric_minute", indexRuleBinding.getSubject().getName());
+ //test data query
+ Instant now = Instant.now();
+ Instant begin = now.minus(15, ChronoUnit.MINUTES);
+ MeasureWrite measureWrite = client.createMeasureWrite(groupName, "testMetric_minute", now.toEpochMilli());
+ measureWrite.tag("storage-only", "service_id", TagAndValue.stringTagValue("service1"))
+ .tag("searchable", "tag", TagAndValue.stringTagValue("tag1"))
+ .field("value", TagAndValue.longFieldValue(100));
+ CompletableFuture<Void> f = processor.add(measureWrite);
+ f.exceptionally(exp -> {
+ Assertions.fail(exp.getMessage());
+ return null;
+ });
+ f.get(10, TimeUnit.SECONDS);
+
+ MeasureQuery query = new MeasureQuery(Lists.newArrayList(groupName), "testMetric_minute",
+ new TimestampRange(
+ begin.toEpochMilli(),
+ now.plus(1, ChronoUnit.MINUTES).toEpochMilli()
+ ), ImmutableMap.of("service_id", "storage-only", "tag", "searchable"),
+ ImmutableSet.of("value")
+ );
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ MeasureQueryResponse resp = client.query(query);
+ assertNotNull(resp);
+ assertEquals(1, resp.getDataPoints().size());
+ assertEquals("service1", resp.getDataPoints().get(0).getTagValue("service_id"));
+ assertEquals("tag1", resp.getDataPoints().get(0).getTagValue("tag"));
+ assertEquals(100, (Long) resp.getDataPoints().get(0).getFieldValue("value"));
+ });
+
+ // StorageModels.add now dedupes by (name, downsampling); evict the original
+ // registration so the UpdateTestMetric registration takes effect.
+ models.remove(TestMetric.class, StorageManipulationOpt.withSchemaChange());
+ Model updatedModel = models.add(UpdateTestMetric.class, DefaultScopeDefine.SERVICE,
+ new Storage("testMetric", true, DownSampling.Minute),
+ StorageManipulationOpt.withSchemaChange()
+ );
+ config.getMetricsMin().setShardNum(config.getMetricsMin().getShardNum() + 1);
+ config.getMetricsMin().setSegmentInterval(config.getMetricsMin().getSegmentInterval() + 2);
+ config.getMetricsMin().setTtl(config.getMetricsMin().getTtl() + 3);
+ BanyanDBIndexInstaller newInstaller = new BanyanDBIndexInstaller(client, moduleManager, config);
+ newInstaller.isExists(updatedModel, StorageManipulationOpt.withSchemaChange());
+ //test Group update — assert the live group now reflects the values we set on config,
+ //rather than hard-coding the post-mutation numbers (which would couple the test to the
+ //defaults shipped in bydb.yml).
+ BanyandbCommon.Group updatedGroup = client.client.findGroup(groupName);
+ assertEquals(config.getMetricsMin().getShardNum(), updatedGroup.getResourceOpts().getShardNum());
+ assertEquals(config.getMetricsMin().getSegmentInterval(), updatedGroup.getResourceOpts().getSegmentInterval().getNum());
+ assertEquals(config.getMetricsMin().getTtl(), updatedGroup.getResourceOpts().getTtl().getNum());
+ //test Measure update
+ BanyandbDatabase.Measure updatedMeasure = client.client.findMeasure(groupName, "testMetric_minute");
+ assertEquals("storage-only", updatedMeasure.getTagFamilies(0).getName());
+ assertEquals("service_id", updatedMeasure.getTagFamilies(0).getTags(0).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, updatedMeasure.getTagFamilies(0).getTags(0).getType());
+ assertEquals("searchable", updatedMeasure.getTagFamilies(1).getName());
+ assertEquals("tag", updatedMeasure.getTagFamilies(1).getTags(0).getName());
+ assertEquals("new_tag", updatedMeasure.getTagFamilies(1).getTags(1).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, updatedMeasure.getTagFamilies(1).getTags(0).getType());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, updatedMeasure.getTagFamilies(1).getTags(1).getType());
+ assertEquals("service_id", updatedMeasure.getEntity().getTagNames(0));
+ assertEquals("value", updatedMeasure.getFields(0).getName());
+ assertEquals(BanyandbDatabase.FieldType.FIELD_TYPE_INT, updatedMeasure.getFields(0).getFieldType());
+ assertEquals("new_value", updatedMeasure.getFields(1).getName());
+ assertEquals(BanyandbDatabase.FieldType.FIELD_TYPE_INT, updatedMeasure.getFields(1).getFieldType());
+ //test IndexRule update
+ BanyandbDatabase.IndexRule updatedIndexRuleTag = client.client.findIndexRule(groupName, "tag");
+ assertEquals("", updatedIndexRuleTag.getAnalyzer());
+ assertFalse(updatedIndexRuleTag.getNoSort());
+ BanyandbDatabase.IndexRule updatedIndexRuleNewTag = client.client.findIndexRule(groupName, "new_tag");
+ assertTrue(updatedIndexRuleNewTag.getNoSort());
+ //test IndexRuleBinding update
+ BanyandbDatabase.IndexRuleBinding updatedIndexRuleBinding = client.client.findIndexRuleBinding(
+ groupName, "testMetric_minute");
+ assertEquals("tag", updatedIndexRuleBinding.getRules(0));
+ assertEquals("new_tag", updatedIndexRuleBinding.getRules(1));
+ assertEquals("testMetric_minute", updatedIndexRuleBinding.getSubject().getName());
+ //test data
+ MeasureWrite updatedMeasureWrite = client.createMeasureWrite(groupName, "testMetric_minute", now.plus(10, ChronoUnit.MINUTES).toEpochMilli());
+ updatedMeasureWrite.tag("storage-only", "service_id", TagAndValue.stringTagValue("service2"))
+ .tag("searchable", "tag", TagAndValue.stringTagValue("tag1"))
+ .tag("searchable", "new_tag", TagAndValue.stringTagValue("new_tag1"))
+ .field("value", TagAndValue.longFieldValue(101))
+ .field("new_value", TagAndValue.longFieldValue(1000));
+ CompletableFuture<Void> cf = processor.add(updatedMeasureWrite);
+ cf.exceptionally(exp -> {
+ Assertions.fail(exp.getMessage());
+ return null;
+ });
+ cf.get(10, TimeUnit.SECONDS);
+ MeasureQuery updatedQuery = new MeasureQuery(
+ Lists.newArrayList(groupName), "testMetric_minute",
+ new TimestampRange(begin.toEpochMilli(), now.plus(15, ChronoUnit.MINUTES).toEpochMilli()),
+ ImmutableMap.of("service_id", "storage-only", "tag", "searchable", "new_tag", "searchable"),
+ ImmutableSet.of("value", "new_value")
+ );
+ await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+ MeasureQueryResponse updatedResp = client.query(updatedQuery);
+ assertNotNull(updatedResp);
+ assertEquals(2, updatedResp.getDataPoints().size());
+ // Index by service_id so the assertions don't depend on server-side ordering
+ // (MeasureQuery doesn't set an orderBy and BanyanDB result order is not contractually stable).
+ Map<String, DataPoint> byService = updatedResp.getDataPoints().stream()
+ .collect(Collectors.toMap(dp -> (String) dp.getTagValue("service_id"), dp -> dp));
+ DataPoint dp1 = byService.get("service1");
+ assertNotNull(dp1);
+ assertEquals("tag1", dp1.getTagValue("tag"));
+ assertEquals(100, (Long) dp1.getFieldValue("value"));
+ DataPoint dp2 = byService.get("service2");
+ assertNotNull(dp2);
+ assertEquals("tag1", dp2.getTagValue("tag"));
+ assertEquals("new_tag1", dp2.getTagValue("new_tag"));
+ assertEquals(101, (Long) dp2.getFieldValue("value"));
+ assertEquals(1000, (Long) dp2.getFieldValue("new_value"));
+ });
+ }
+
+ @Test
+ public void testStreamInstallAndUpdate() throws Exception {
+ DownSamplingConfigService downSamplingConfigService = new DownSamplingConfigService(Arrays.asList("minute"));
+ ModuleManager moduleManager = mock(ModuleManager.class);
+ ModuleProviderHolder moduleProviderHolder = mock(ModuleProviderHolder.class);
+ ModuleServiceHolder moduleServiceHolder = mock(ModuleServiceHolder.class);
+ when(moduleManager.find(CoreModule.NAME)).thenReturn(moduleProviderHolder);
+ when(moduleProviderHolder.provider()).thenReturn(moduleServiceHolder);
+ when(moduleServiceHolder.getService(DownSamplingConfigService.class)).thenReturn(downSamplingConfigService);
+
+ StorageModels models = new StorageModels();
+ Model model = models.add(TestStream.class, DefaultScopeDefine.SERVICE,
+ new Storage("testStream", true, DownSampling.Second),
+ StorageManipulationOpt.withSchemaChange()
+ );
+ BanyanDBIndexInstaller installer = new BanyanDBIndexInstaller(client, moduleManager, config);
+ installer.isExists(model, StorageManipulationOpt.withSchemaChange());
+ // test Group install
+ String groupName = MetadataRegistry.convertGroupName(
+ config.getGlobal().getNamespace(),
+ BanyanDB.StreamGroup.RECORDS_LOG.getName()
+ );
+ BanyandbCommon.Group group = client.client.findGroup(groupName);
+ assertEquals(BanyandbCommon.Catalog.CATALOG_STREAM, group.getCatalog());
+ assertEquals(config.getRecordsLog().getSegmentInterval(), group.getResourceOpts().getSegmentInterval().getNum());
+ assertEquals(config.getRecordsLog().getShardNum(), group.getResourceOpts().getShardNum());
+ assertEquals(BanyandbCommon.IntervalRule.Unit.UNIT_DAY, group.getResourceOpts().getSegmentInterval().getUnit());
+ assertEquals(config.getRecordsLog().getTtl(), group.getResourceOpts().getTtl().getNum());
+ assertEquals(BanyandbCommon.IntervalRule.Unit.UNIT_DAY, group.getResourceOpts().getTtl().getUnit());
+
+ installer.createTable(model);
+ // test Stream install
+ BanyandbDatabase.Stream stream = client.client.findStream(groupName, "testStream");
+ assertEquals("storage-only", stream.getTagFamilies(0).getName());
+ assertEquals("service_id", stream.getTagFamilies(0).getTags(0).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, stream.getTagFamilies(0).getTags(0).getType());
+ assertEquals("timestamp", stream.getTagFamilies(0).getTags(1).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_INT, stream.getTagFamilies(0).getTags(1).getType());
+ assertEquals("searchable", stream.getTagFamilies(1).getName());
+ assertEquals("tag", stream.getTagFamilies(1).getTags(0).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, stream.getTagFamilies(1).getTags(0).getType());
+ assertEquals("service_id", stream.getEntity().getTagNames(0));
+ // test IndexRule install
+ BanyandbDatabase.IndexRule indexRuleTag = client.client.findIndexRule(groupName, "tag");
+ assertEquals("url", indexRuleTag.getAnalyzer());
+ assertTrue(indexRuleTag.getNoSort());
+ // test IndexRuleBinding install
+ BanyandbDatabase.IndexRuleBinding indexRuleBinding = client.client.findIndexRuleBinding(
+ groupName, "testStream");
+ assertEquals("tag", indexRuleBinding.getRules(0));
+ assertEquals("testStream", indexRuleBinding.getSubject().getName());
+
+ // StorageModels.add now dedupes by (name, downsampling); evict the original
+ // registration so the UpdateTestStream registration takes effect.
+ models.remove(TestStream.class, StorageManipulationOpt.withSchemaChange());
+ Model updatedModel = models.add(UpdateTestStream.class, DefaultScopeDefine.SERVICE,
+ new Storage("testStream", true, DownSampling.Second),
+ StorageManipulationOpt.withSchemaChange()
+ );
+ BanyanDBIndexInstaller newInstaller = new BanyanDBIndexInstaller(client, moduleManager, config);
+ newInstaller.isExists(updatedModel, StorageManipulationOpt.withSchemaChange());
+ // test Stream update
+ BanyandbDatabase.Stream updatedStream = client.client.findStream(groupName, "testStream");
+ assertEquals("storage-only", updatedStream.getTagFamilies(0).getName());
+ assertEquals("service_id", updatedStream.getTagFamilies(0).getTags(0).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, updatedStream.getTagFamilies(0).getTags(0).getType());
+ assertEquals("timestamp", updatedStream.getTagFamilies(0).getTags(1).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_INT, updatedStream.getTagFamilies(0).getTags(1).getType());
+ assertEquals("searchable", updatedStream.getTagFamilies(1).getName());
+ assertEquals("tag", updatedStream.getTagFamilies(1).getTags(0).getName());
+ assertEquals("new_tag", updatedStream.getTagFamilies(1).getTags(1).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, updatedStream.getTagFamilies(1).getTags(0).getType());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, updatedStream.getTagFamilies(1).getTags(1).getType());
+ assertEquals("service_id", updatedStream.getEntity().getTagNames(0));
+ // test IndexRule update
+ BanyandbDatabase.IndexRule updatedIndexRuleTag = client.client.findIndexRule(groupName, "tag");
+ assertEquals("", updatedIndexRuleTag.getAnalyzer());
+ assertFalse(updatedIndexRuleTag.getNoSort());
+ BanyandbDatabase.IndexRule updatedIndexRuleNewTag = client.client.findIndexRule(groupName, "new_tag");
+ assertTrue(updatedIndexRuleNewTag.getNoSort());
+ // test IndexRuleBinding update
+ BanyandbDatabase.IndexRuleBinding updatedIndexRuleBinding = client.client.findIndexRuleBinding(
+ groupName, "testStream");
+ assertEquals("tag", updatedIndexRuleBinding.getRules(0));
+ assertEquals("new_tag", updatedIndexRuleBinding.getRules(1));
+ assertEquals("testStream", updatedIndexRuleBinding.getSubject().getName());
+ }
+
+ /**
+ * Drive the allowBootReshape path: install {@link TestStream} first under
+ * {@link StorageManipulationOpt#withSchemaChange()} so the stream lives on the backend
+ * with the original shape, then re-install a class that adds {@code new_tag} and opts in
+ * via {@code allowBootReshape = true} under {@link StorageManipulationOpt#schemaCreateIfAbsent()}.
+ * The installer should apply the additive update during boot rather than recording
+ * {@link StorageManipulationOpt.Outcome#SKIPPED_SHAPE_MISMATCH}.
+ */
+ @Test
+ public void testStreamAdditiveBootReshape() throws Exception {
+ DownSamplingConfigService downSamplingConfigService = new DownSamplingConfigService(Arrays.asList("minute"));
+ ModuleManager moduleManager = mock(ModuleManager.class);
+ ModuleProviderHolder moduleProviderHolder = mock(ModuleProviderHolder.class);
+ ModuleServiceHolder moduleServiceHolder = mock(ModuleServiceHolder.class);
+ when(moduleManager.find(CoreModule.NAME)).thenReturn(moduleProviderHolder);
+ when(moduleProviderHolder.provider()).thenReturn(moduleServiceHolder);
+ when(moduleServiceHolder.getService(DownSamplingConfigService.class)).thenReturn(downSamplingConfigService);
+
+ StorageModels models = new StorageModels();
+ Model baseModel = models.add(TestStream.class, DefaultScopeDefine.SERVICE,
+ new Storage("testStream", true, DownSampling.Second),
+ StorageManipulationOpt.withSchemaChange());
+ BanyanDBIndexInstaller installer = new BanyanDBIndexInstaller(client, moduleManager, config);
+ installer.isExists(baseModel, StorageManipulationOpt.withSchemaChange());
+ installer.createTable(baseModel);
+
+ String groupName = MetadataRegistry.convertGroupName(
+ config.getGlobal().getNamespace(), BanyanDB.StreamGroup.RECORDS_LOG.getName());
+ BanyandbDatabase.Stream initial = client.client.findStream(groupName, "testStream");
+ assertEquals(1, initial.getTagFamilies(1).getTagsCount());
+
+ models.remove(TestStream.class, StorageManipulationOpt.withSchemaChange());
+ Model reshapedModel = models.add(TestStreamAdditiveReshapeOn.class, DefaultScopeDefine.SERVICE,
+ new Storage("testStream", true, DownSampling.Second),
+ StorageManipulationOpt.withSchemaChange());
+ assertTrue(reshapedModel.isAllowBootReshape());
+
+ StorageManipulationOpt bootOpt = StorageManipulationOpt.schemaCreateIfAbsent();
+ new BanyanDBIndexInstaller(client, moduleManager, config).isExists(reshapedModel, bootOpt);
+
+ BanyandbDatabase.Stream reshaped = client.client.findStream(groupName, "testStream");
+ assertEquals(2, reshaped.getTagFamilies(1).getTagsCount());
+ assertEquals("new_tag", reshaped.getTagFamilies(1).getTags(1).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, reshaped.getTagFamilies(1).getTags(1).getType());
+
+ boolean updatedRecorded = bootOpt.getOutcomes().stream()
+ .anyMatch(o -> "stream".equals(o.getResourceType())
+ && "testStream".equals(o.getResourceName())
+ && o.getStatus() == StorageManipulationOpt.Outcome.UPDATED);
+ assertTrue(updatedRecorded, "expected UPDATED outcome for additive boot reshape, got " + bootOpt.getOutcomes());
+ }
+
+ /**
+ * Same setup as {@link #testStreamAdditiveBootReshape} but the re-install class leaves
+ * {@code allowBootReshape} at its default ({@code false}). Boot must refuse to reshape
+ * even though the diff is additive — record {@link StorageManipulationOpt.Outcome#SKIPPED_SHAPE_MISMATCH}
+ * and leave the live stream untouched.
+ */
+ @Test
+ public void testStreamAdditiveBootReshape_optOutSkips() throws Exception {
+ DownSamplingConfigService downSamplingConfigService = new DownSamplingConfigService(Arrays.asList("minute"));
+ ModuleManager moduleManager = mock(ModuleManager.class);
+ ModuleProviderHolder moduleProviderHolder = mock(ModuleProviderHolder.class);
+ ModuleServiceHolder moduleServiceHolder = mock(ModuleServiceHolder.class);
+ when(moduleManager.find(CoreModule.NAME)).thenReturn(moduleProviderHolder);
+ when(moduleProviderHolder.provider()).thenReturn(moduleServiceHolder);
+ when(moduleServiceHolder.getService(DownSamplingConfigService.class)).thenReturn(downSamplingConfigService);
+
+ StorageModels models = new StorageModels();
+ Model baseModel = models.add(TestStream.class, DefaultScopeDefine.SERVICE,
+ new Storage("testStream", true, DownSampling.Second),
+ StorageManipulationOpt.withSchemaChange());
+ BanyanDBIndexInstaller installer = new BanyanDBIndexInstaller(client, moduleManager, config);
+ installer.isExists(baseModel, StorageManipulationOpt.withSchemaChange());
+ installer.createTable(baseModel);
+
+ models.remove(TestStream.class, StorageManipulationOpt.withSchemaChange());
+ Model reshapedModel = models.add(TestStreamAdditiveReshapeOff.class, DefaultScopeDefine.SERVICE,
+ new Storage("testStream", true, DownSampling.Second),
+ StorageManipulationOpt.withSchemaChange());
+ assertFalse(reshapedModel.isAllowBootReshape());
+
+ StorageManipulationOpt bootOpt = StorageManipulationOpt.schemaCreateIfAbsent();
+ new BanyanDBIndexInstaller(client, moduleManager, config).isExists(reshapedModel, bootOpt);
+
+ String groupName = MetadataRegistry.convertGroupName(
+ config.getGlobal().getNamespace(), BanyanDB.StreamGroup.RECORDS_LOG.getName());
+ BanyandbDatabase.Stream live = client.client.findStream(groupName, "testStream");
+ assertEquals(1, live.getTagFamilies(1).getTagsCount());
+
+ boolean skipRecorded = bootOpt.getOutcomes().stream()
+ .anyMatch(o -> "stream".equals(o.getResourceType())
+ && "testStream".equals(o.getResourceName())
+ && o.getStatus() == StorageManipulationOpt.Outcome.SKIPPED_SHAPE_MISMATCH);
+ assertTrue(skipRecorded, "expected SKIPPED_SHAPE_MISMATCH without allowBootReshape opt-in, got " + bootOpt.getOutcomes());
+ }
+
+ /**
+ * Measure variant of {@link #testStreamAdditiveBootReshape}: install {@link TestMetric},
+ * then re-install {@link TestMetricAdditiveReshapeOn} which adds {@code new_tag} and
+ * {@code new_value} (a field) and opts in via {@code allowBootReshape = true}. The boot
+ * installer should apply the additive update and record
+ * {@link StorageManipulationOpt.Outcome#UPDATED}.
+ */
+ @Test
+ public void testMeasureAdditiveBootReshape() throws Exception {
+ DownSamplingConfigService downSamplingConfigService = new DownSamplingConfigService(Arrays.asList("minute"));
+ ModuleManager moduleManager = mock(ModuleManager.class);
+ ModuleProviderHolder moduleProviderHolder = mock(ModuleProviderHolder.class);
+ ModuleServiceHolder moduleServiceHolder = mock(ModuleServiceHolder.class);
+ when(moduleManager.find(CoreModule.NAME)).thenReturn(moduleProviderHolder);
+ when(moduleProviderHolder.provider()).thenReturn(moduleServiceHolder);
+ when(moduleServiceHolder.getService(DownSamplingConfigService.class)).thenReturn(downSamplingConfigService);
+
+ StorageModels models = new StorageModels();
+ Model baseModel = models.add(TestMetric.class, DefaultScopeDefine.SERVICE,
+ new Storage("testMetric", true, DownSampling.Minute),
+ StorageManipulationOpt.withSchemaChange());
+ BanyanDBIndexInstaller installer = new BanyanDBIndexInstaller(client, moduleManager, config);
+ installer.isExists(baseModel, StorageManipulationOpt.withSchemaChange());
+ installer.createTable(baseModel);
+
+ models.remove(TestMetric.class, StorageManipulationOpt.withSchemaChange());
+ Model reshapedModel = models.add(TestMetricAdditiveReshapeOn.class, DefaultScopeDefine.SERVICE,
+ new Storage("testMetric", true, DownSampling.Minute),
+ StorageManipulationOpt.withSchemaChange());
+ assertTrue(reshapedModel.isAllowBootReshape());
+
+ StorageManipulationOpt bootOpt = StorageManipulationOpt.schemaCreateIfAbsent();
+ new BanyanDBIndexInstaller(client, moduleManager, config).isExists(reshapedModel, bootOpt);
+
+ String groupName = MetadataRegistry.convertGroupName(
+ config.getGlobal().getNamespace(), BanyanDB.MeasureGroup.METRICS_MINUTE.getName());
+ BanyandbDatabase.Measure reshaped = client.client.findMeasure(groupName, "testMetric_minute");
+ assertEquals(2, reshaped.getTagFamilies(1).getTagsCount());
+ assertEquals("new_tag", reshaped.getTagFamilies(1).getTags(1).getName());
+ assertEquals(2, reshaped.getFieldsCount());
+ assertEquals("new_value", reshaped.getFields(1).getName());
+
+ boolean updatedRecorded = bootOpt.getOutcomes().stream()
+ .anyMatch(o -> "measure".equals(o.getResourceType())
+ && "testMetric_minute".equals(o.getResourceName())
+ && o.getStatus() == StorageManipulationOpt.Outcome.UPDATED);
+ assertTrue(updatedRecorded, "expected UPDATED outcome for additive measure boot reshape, got " + bootOpt.getOutcomes());
+ }
+
+ /**
+ * Opt-in is necessary but not sufficient: even with {@code allowBootReshape = true} a
+ * diff that is not purely additive (here, the {@code tag} column flips from
+ * {@code String} → {@code long}, i.e. {@code TAG_TYPE_STRING} → {@code TAG_TYPE_INT})
+ * must be refused at boot. {@link StorageManipulationOpt.Outcome#SKIPPED_SHAPE_MISMATCH}
+ * is recorded and the live stream stays unchanged, forcing the operator to do an
+ * explicit drop+recreate (the only safe path for identity-breaking changes).
+ */
+ @Test
+ public void testStreamNonAdditiveBootReshape_optInStillSkips() throws Exception {
+ DownSamplingConfigService downSamplingConfigService = new DownSamplingConfigService(Arrays.asList("minute"));
+ ModuleManager moduleManager = mock(ModuleManager.class);
+ ModuleProviderHolder moduleProviderHolder = mock(ModuleProviderHolder.class);
+ ModuleServiceHolder moduleServiceHolder = mock(ModuleServiceHolder.class);
+ when(moduleManager.find(CoreModule.NAME)).thenReturn(moduleProviderHolder);
+ when(moduleProviderHolder.provider()).thenReturn(moduleServiceHolder);
+ when(moduleServiceHolder.getService(DownSamplingConfigService.class)).thenReturn(downSamplingConfigService);
+
+ StorageModels models = new StorageModels();
+ Model baseModel = models.add(TestStream.class, DefaultScopeDefine.SERVICE,
+ new Storage("testStream", true, DownSampling.Second),
+ StorageManipulationOpt.withSchemaChange());
+ BanyanDBIndexInstaller installer = new BanyanDBIndexInstaller(client, moduleManager, config);
+ installer.isExists(baseModel, StorageManipulationOpt.withSchemaChange());
+ installer.createTable(baseModel);
+
+ models.remove(TestStream.class, StorageManipulationOpt.withSchemaChange());
+ Model reshapedModel = models.add(TestStreamNonAdditiveReshapeOn.class, DefaultScopeDefine.SERVICE,
+ new Storage("testStream", true, DownSampling.Second),
+ StorageManipulationOpt.withSchemaChange());
+ assertTrue(reshapedModel.isAllowBootReshape());
+
+ StorageManipulationOpt bootOpt = StorageManipulationOpt.schemaCreateIfAbsent();
+ new BanyanDBIndexInstaller(client, moduleManager, config).isExists(reshapedModel, bootOpt);
+
+ String groupName = MetadataRegistry.convertGroupName(
+ config.getGlobal().getNamespace(), BanyanDB.StreamGroup.RECORDS_LOG.getName());
+ BanyandbDatabase.Stream live = client.client.findStream(groupName, "testStream");
+ assertEquals("tag", live.getTagFamilies(1).getTags(0).getName());
+ assertEquals(BanyandbDatabase.TagType.TAG_TYPE_STRING, live.getTagFamilies(1).getTags(0).getType());
+
+ boolean skipRecorded = bootOpt.getOutcomes().stream()
+ .anyMatch(o -> "stream".equals(o.getResourceType())
+ && "testStream".equals(o.getResourceName())
+ && o.getStatus() == StorageManipulationOpt.Outcome.SKIPPED_SHAPE_MISMATCH);
+ assertTrue(skipRecorded,
+ "expected SKIPPED_SHAPE_MISMATCH for non-additive (type-change) diff even with opt-in, got " + bootOpt.getOutcomes());
+ }
+
+ /**
+ * Non-additive variant: {@code tag} is now {@code long} (TAG_TYPE_INT) where the live
+ * stream has it as {@code String} (TAG_TYPE_STRING). Boot must refuse to reshape even
+ * with {@code allowBootReshape = true}.
+ */
+ @Stream(name = "testStream", scopeId = DefaultScopeDefine.SERVICE,
+ builder = TestStreamNonAdditiveReshapeOn.Builder.class, processor = RecordStreamProcessor.class,
+ allowBootReshape = true)
+ @BanyanDB.Group(streamGroup = BanyanDB.StreamGroup.RECORDS_LOG)
+ @BanyanDB.TimestampColumn("timestamp")
+ private static class TestStreamNonAdditiveReshapeOn extends Record {
+ @Column(name = "service_id")
+ @BanyanDB.SeriesID(index = 0)
+ private String serviceId;
+ @Column(name = "tag")
+ private long tag;
+ @Column(name = "timestamp")
+ private long timestamp;
+
+ @Override
+ public StorageID id() {
+ return new StorageID();
+ }
+
+ static class Builder implements StorageBuilder<StorageData> {
+ @Override
+ public StorageData storage2Entity(final Convert2Entity converter) {
+ return null;
+ }
+
+ @Override
+ public void entity2Storage(final StorageData entity, final Convert2Storage converter) {
+ }
+ }
+ }
+
+ /**
+ * Mirror of {@link UpdateTestStream} but trimmed to a purely-additive shape change
+ * (only {@code new_tag} is added; {@code tag}'s index settings stay matched to
+ * {@link TestStream}) and with the new {@code allowBootReshape} opt-in flipped on.
+ */
+ @Stream(name = "testStream", scopeId = DefaultScopeDefine.SERVICE,
+ builder = TestStreamAdditiveReshapeOn.Builder.class, processor = RecordStreamProcessor.class,
+ allowBootReshape = true)
+ @BanyanDB.Group(streamGroup = BanyanDB.StreamGroup.RECORDS_LOG)
+ @BanyanDB.TimestampColumn("timestamp")
+ private static class TestStreamAdditiveReshapeOn extends Record {
+ @Column(name = "service_id")
+ @BanyanDB.SeriesID(index = 0)
+ private String serviceId;
+ @Column(name = "tag")
+ @BanyanDB.MatchQuery(analyzer = BanyanDB.MatchQuery.AnalyzerType.URL)
+ private String tag;
+ @Column(name = "new_tag")
+ private String newTag;
+ @Column(name = "timestamp")
+ private long timestamp;
+
+ @Override
+ public StorageID id() {
+ return new StorageID();
+ }
+
+ static class Builder implements StorageBuilder<StorageData> {
+ @Override
+ public StorageData storage2Entity(final Convert2Entity converter) {
+ return null;
+ }
+
+ @Override
+ public void entity2Storage(final StorageData entity, final Convert2Storage converter) {
+ }
+ }
+ }
+
+ /**
+ * Same additive shape as {@link TestStreamAdditiveReshapeOn} but without the opt-in —
+ * boot must refuse to reshape and record SKIPPED_SHAPE_MISMATCH.
+ */
+ @Stream(name = "testStream", scopeId = DefaultScopeDefine.SERVICE,
+ builder = TestStreamAdditiveReshapeOff.Builder.class, processor = RecordStreamProcessor.class)
+ @BanyanDB.Group(streamGroup = BanyanDB.StreamGroup.RECORDS_LOG)
+ @BanyanDB.TimestampColumn("timestamp")
+ private static class TestStreamAdditiveReshapeOff extends Record {
+ @Column(name = "service_id")
+ @BanyanDB.SeriesID(index = 0)
+ private String serviceId;
+ @Column(name = "tag")
+ @BanyanDB.MatchQuery(analyzer = BanyanDB.MatchQuery.AnalyzerType.URL)
+ private String tag;
+ @Column(name = "new_tag")
+ private String newTag;
+ @Column(name = "timestamp")
+ private long timestamp;
+
+ @Override
+ public StorageID id() {
+ return new StorageID();
+ }
+
+ static class Builder implements StorageBuilder<StorageData> {
+ @Override
+ public StorageData storage2Entity(final Convert2Entity converter) {
+ return null;
+ }
+
+ @Override
+ public void entity2Storage(final StorageData entity, final Convert2Storage converter) {
+ }
+ }
+ }
+
+ /**
+ * Mirror of {@link UpdateTestMetric} but trimmed to a purely-additive shape change
+ * (new tag, new field) and with {@code allowBootReshape = true}.
+ */
+ @Stream(name = "testMetric", scopeId = DefaultScopeDefine.SERVICE,
+ builder = TestMetricAdditiveReshapeOn.Builder.class, processor = MetricsStreamProcessor.class,
+ allowBootReshape = true)
+ private static class TestMetricAdditiveReshapeOn {
+ @Column(name = "service_id")
+ @BanyanDB.SeriesID(index = 0)
+ @BanyanDB.ShardingKey(index = 0)
+ private String serviceId;
+ @Column(name = "tag")
+ @BanyanDB.MatchQuery(analyzer = BanyanDB.MatchQuery.AnalyzerType.URL)
+ private String tag;
+ @Column(name = "new_tag")
+ private String newTag;
+ @Column(name = "value", dataType = Column.ValueDataType.COMMON_VALUE)
+ @BanyanDB.MeasureField
+ private long value;
+ @Column(name = "new_value", storageOnly = true)
+ @BanyanDB.MeasureField
+ private long newValue;
+
+ static class Builder implements StorageBuilder<StorageData> {
+ @Override
+ public StorageData storage2Entity(final Convert2Entity converter) {
+ return null;
+ }
+
+ @Override
+ public void entity2Storage(final StorageData entity, final Convert2Storage converter) {
+ }
+ }
+ }
+
+ @Stream(name = "testStream", scopeId = DefaultScopeDefine.SERVICE,
+ builder = TestStream.Builder.class, processor = RecordStreamProcessor.class)
+ @BanyanDB.Group(streamGroup = BanyanDB.StreamGroup.RECORDS_LOG)
+ @BanyanDB.TimestampColumn("timestamp")
+ private static class TestStream extends Record {
+ @Column(name = "service_id")
+ @BanyanDB.SeriesID(index = 0)
+ private String serviceId;
+ @Column(name = "tag")
+ @BanyanDB.MatchQuery(analyzer = BanyanDB.MatchQuery.AnalyzerType.URL)
+ private String tag;
+ @Column(name = "timestamp")
+ private long timestamp;
+
+ @Override
+ public StorageID id() {
+ return new StorageID();
+ }
+
+ static class Builder implements StorageBuilder<StorageData> {
+ @Override
+ public StorageData storage2Entity(final Convert2Entity converter) {
+ return null;
+ }
+
+ @Override
+ public void entity2Storage(final StorageData entity, final Convert2Storage converter) {
+
+ }
+ }
+ }
+
+ @Stream(name = "testStream", scopeId = DefaultScopeDefine.SERVICE,
+ builder = UpdateTestStream.Builder.class, processor = RecordStreamProcessor.class)
+ @BanyanDB.Group(streamGroup = BanyanDB.StreamGroup.RECORDS_LOG)
+ @BanyanDB.TimestampColumn("timestamp")
+ private static class UpdateTestStream extends Record {
+ @Column(name = "service_id")
+ @BanyanDB.SeriesID(index = 0)
+ private String serviceId;
+ @Column(name = "tag")
+ @BanyanDB.EnableSort
+ private String tag;
+ @Column(name = "new_tag")
+ private String newTag;
+ @Column(name = "timestamp")
+ private long timestamp;
+
+ @Override
+ public StorageID id() {
+ return new StorageID();
+ }
+
+ static class Builder implements StorageBuilder<StorageData> {
+ @Override
+ public StorageData storage2Entity(final Convert2Entity converter) {
+ return null;
+ }
+
+ @Override
+ public void entity2Storage(final StorageData entity, final Convert2Storage converter) {
+
+ }
+ }
+ }
+
+ @Stream(name = "testMetric", scopeId = DefaultScopeDefine.SERVICE,
+ builder = TestMetric.Builder.class, processor = MetricsStreamProcessor.class)
+ private static class TestMetric {
+ @Column(name = "service_id")
+ @BanyanDB.SeriesID(index = 0)
+ @BanyanDB.ShardingKey(index = 0)
+ private String serviceId;
+ @Column(name = "tag")
+ @BanyanDB.MatchQuery(analyzer = BanyanDB.MatchQuery.AnalyzerType.URL)
+ private String tag;
+ @Column(name = "value", dataType = Column.ValueDataType.COMMON_VALUE)
+ @BanyanDB.MeasureField
+ private long value;
+
+ static class Builder implements StorageBuilder<StorageData> {
+ @Override
+ public StorageData storage2Entity(final Convert2Entity converter) {
+ return null;
+ }
+
+ @Override
+ public void entity2Storage(final StorageData entity, final Convert2Storage converter) {
+
+ }
+ }
+ }
+
+ @Stream(name = "testMetric", scopeId = DefaultScopeDefine.SERVICE,
+ builder = UpdateTestMetric.Builder.class, processor = MetricsStreamProcessor.class)
+ private static class UpdateTestMetric {
+ @Column(name = "service_id")
+ @BanyanDB.SeriesID(index = 0)
+ private String serviceId;
+ @Column(name = "tag")
+ @BanyanDB.EnableSort
+ private String tag;
+ @Column(name = "new_tag")
+ private String newTag;
+ @Column(name = "value", dataType = Column.ValueDataType.COMMON_VALUE)
+ @BanyanDB.MeasureField
+ private long value;
+ @Column(name = "new_value", storageOnly = true)
+ @BanyanDB.MeasureField
+ private long newValue;
+
+ static class Builder implements StorageBuilder<StorageData> {
+ @Override
+ public StorageData storage2Entity(final Convert2Entity converter) {
+ return null;
+ }
+
+ @Override
+ public void entity2Storage(final StorageData entity, final Convert2Storage converter) {
+
+ }
+ }
+ }
+}
diff --git a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/resources/bydb.yml b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/resources/bydb.yml
index 1260ffa34d..b1ce48367b 100644
--- a/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/resources/bydb.yml
+++ b/oap-server/server-storage-plugin/storage-banyandb-plugin/src/test/resources/bydb.yml
@@ -43,6 +43,7 @@ global:
# The batch size for querying profile data.
profileDataQueryBatchSize: ${SW_STORAGE_BANYANDB_QUERY_PROFILE_DATA_BATCH_SIZE:100}
asyncProfilerTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_ASYNC_PROFILER_TASK_QUERY_MAX_SIZE:200}
+ pprofTaskQueryMaxSize: ${SW_STORAGE_BANYANDB_PPROF_TASK_QUERY_MAX_SIZE:200}
user: ${SW_STORAGE_BANYANDB_USER:""}
password: ${SW_STORAGE_BANYANDB_PASSWORD:""}
# If the BanyanDB server is configured with TLS, configure the TLS cert file path and enable TLS connection.
@@ -52,6 +53,9 @@ global:
# The namespace in BanyanDB to store the data of OAP, if not set, the default is "sw".
# OAP will create BanyanDB Groups using the format of "{namespace}_{group name}", such as "sw_records".
namespace: ${SW_NAMESPACE:"sw"}
+ # The compatible server API versions of BanyanDB.
+ # The compatible BanyanDB Server version number can be found via the [API versions mapping](https://skywalking.apache.org/docs/skywalking-banyandb/latest/installation/versions/).
+ compatibleServerApiVersions: ${SW_STORAGE_BANYANDB_COMPATIBLE_SERVER_API_VERSIONS:"0.10"}
groups:
# The group settings of record.
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTest.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTest.java
index 931edd6615..0002de624b 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTest.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/base/TimeSeriesUtilsTest.java
@@ -43,15 +43,15 @@ public class TimeSeriesUtilsTest {
@BeforeEach
public void prepare() {
superDatasetModel = new Model("superDatasetModel", Lists.newArrayList(),
- 0, DownSampling.Second, true, Record.class, true,
+ 0, DownSampling.Second, true, Record.class, true, false,
new SQLDatabaseModelExtension(), new BanyanDBModelExtension(), new ElasticSearchModelExtension()
);
normalRecordModel = new Model("normalRecordModel", Lists.newArrayList(),
- 0, DownSampling.Second, false, Record.class, true,
+ 0, DownSampling.Second, false, Record.class, true, false,
new SQLDatabaseModelExtension(), new BanyanDBModelExtension(), new ElasticSearchModelExtension()
);
normalMetricsModel = new Model("normalMetricsModel", Lists.newArrayList(),
- 0, DownSampling.Minute, false, Metrics.class, true,
+ 0, DownSampling.Minute, false, Metrics.class, true, false,
new SQLDatabaseModelExtension(), new BanyanDBModelExtension(), new ElasticSearchModelExtension()
);
TimeSeriesUtils.setSUPER_DATASET_DAY_STEP(1);
diff --git a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCHistoryDeleteDAOIT.java b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCHistoryDeleteDAOIT.java
index c0a729ba29..ecbc900913 100644
--- a/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCHistoryDeleteDAOIT.java
+++ b/oap-server/server-storage-plugin/storage-jdbc-hikaricp-plugin/src/test/java/org/apache/skywalking/oap/server/storage/plugin/jdbc/common/dao/JDBCHistoryDeleteDAOIT.java
@@ -107,7 +107,7 @@ class JDBCHistoryDeleteDAOIT {
new SQLDatabaseExtension(), null, null),
new ModelColumn(new ColumnName(timeBucketColumn), Long.class, Long.class, false, false, false, 0,
new SQLDatabaseExtension(), null, null)
- ), 1, DownSampling.Minute, false, ServiceTraffic.class, false, new SQLDatabaseModelExtension(),
+ ), 1, DownSampling.Minute, false, ServiceTraffic.class, false, false, new SQLDatabaseModelExtension(),
new BanyanDBModelExtension(), new ElasticSearchModelExtension());
TableMetaInfo.addModel(model);