feat(tsdb): land s16 tiered retention and rollup — transparent tier reads on the shared caggs

The timescale adapter now owns the S16 lifecycle: raw 30 days (seed 05, was 180) -> 1-minute tier 1 year (configurable via dc3.tsdb.timescale.rollup.minute-keep-days) -> 1-hour tier forever. bucketedAggregate and bucketedCount serve bucket widths at or above a tier's granularity from that tier with exactly composable expressions (AVG recombines as SUM(num_sum)/SUM(num_count), COUNT as SUM(sample_count)); PERCENTILE and FIRST/LAST stay on the raw path — the shared caggs carry textual first/last only, and tiered numeric first/last would version-skew deployments created before such columns existed. Single-window aggregate() stays raw by design: alarm windows must never read through materialization lag. The tiers are the observability pipeline's existing real-time continuous aggregates (cagg_point_value_1m/1h, queried by Grafana) — one structure serves both consumers; the adapter bootstraps the same names with an identical shape via IF NOT EXISTS, so embedded seed boots, standalone adapter boots and already-running deployments all converge on it without rebuilding anything. Real-time mode (materialized_only=FALSE, no longer the TS 2.13+ default) keeps tier reads correct immediately after an append; the refresh policies only move aggregation work off the read path. deleteRange refreshes both tiers afterwards. Also fixed along the way: TDengine's bucketedAggregate PERCENTILE hit the same "single table query" limit as aggregate() and now routes per-series subtables. Gates: timescale TCK 24/24 including the new rollup case (tiered COUNT/AVG/LAST/bucketedCount and bucketed P50 all verified against raw scans before any refresh materialized); TDengine TCK 24/24 with the same case passing via honest raw degradation (2 declared-capability skips).
2026-09-01 15:33:10 +08:00 · 2026-08-21 13:01:45 +08:00
parent 8163aefde6
commit 284c8187bf
7 changed files with 313 additions and 6 deletions
@@ -340,6 +340,57 @@ public abstract class AbstractTsdbContractTest {
        });
    }

+
+    @Test
+    void rollupTierReadsStayConsistentWithRawScans() {
+        SeriesKey key = new SeriesKey(900025, freshId(), freshId());
+        // Five minute-aligned minutes of 10-second samples with distinct values.
+        Instant base = Instant.parse("2026-08-21T02:00:00Z");
+        List<PointValueSample> batch = new ArrayList<>();
+        for (int i = 0; i < 30; i++) {
+            batch.add(sample(key, base.plusSeconds(10L * i), i, 1));
+        }
+        store().append(batch);
+        TimeWindow window = window(base, Duration.ofMinutes(5));
+
+        // COUNT via (possibly tiered) minute buckets must equal the raw count.
+        long tierCount = store().bucketedAggregate(SeriesFilter.of(key), AggregateFunction.COUNT,
+                window, Duration.ofMinutes(1), null, DEADLINE)
+                .getOrDefault(key, List.of()).stream().mapToLong(BucketAggregate::sampleCount).sum();
+        assertThat(tierCount).isEqualTo(store().count(SeriesFilter.of(key), window, DEADLINE));
+
+        // Coarse AVG recombined from tier sums must equal the raw window average.
+        BucketAggregate coarse = store().bucketedAggregate(SeriesFilter.of(key), AggregateFunction.AVG,
+                window, Duration.ofMinutes(5), null, DEADLINE).get(key).getFirst();
+        WindowAggregate rawAvg = store().aggregate(SeriesFilter.of(key), AggregateFunction.AVG,
+                window, null, DEADLINE).get(key);
+        assertThat(coarse.value()).isCloseTo(rawAvg.value(), org.assertj.core.data.Offset.offset(1e-9));
+        assertThat(coarse.sampleCount()).isEqualTo(rawAvg.sampleCount());
+
+        // Coarse LAST recombined from tier lasts must equal the raw window last.
+        BucketAggregate coarseLast = store().bucketedAggregate(SeriesFilter.of(key), AggregateFunction.LAST,
+                window, Duration.ofMinutes(5), null, DEADLINE).get(key).getFirst();
+        WindowAggregate rawLast = store().aggregate(SeriesFilter.of(key), AggregateFunction.LAST,
+                window, null, DEADLINE).get(key);
+        assertThat(coarseLast.value()).isEqualTo(rawLast.value());
+
+        // Tenant-wide bucketed count agrees with the raw tenant count.
+        long tenantBuckets = store().bucketedCount(key.tenantId(), window, Duration.ofMinutes(1), DEADLINE)
+                .stream().mapToLong(BucketAggregate::sampleCount).sum();
+        assertThat(tenantBuckets).isEqualTo(store().count(SeriesFilter.tenantWide(key.tenantId()),
+                window, DEADLINE));
+
+        // PERCENTILE never uses tiers; stores that support it must serve bucketed
+        // percentiles on the raw path without supertable-style failures.
+        if (store().capabilities().percentile()) {
+            List<BucketAggregate> p50 = store().bucketedAggregate(SeriesFilter.of(key),
+                    AggregateFunction.PERCENTILE, window, Duration.ofMinutes(1), 0.5, DEADLINE)
+                    .getOrDefault(key, List.of());
+            assertThat(p50).hasSize(5);
+            assertThat(p50).allSatisfy(bucket -> assertThat(bucket.value()).isNotNull());
+        }
+    }
+
    @Test
    void lastSeenPerSeriesReportsNewestSample() {
        Assumptions.assumeTrue(store().capabilities().tenantWideAnalytics(),
@@ -317,6 +317,35 @@ public final class TdengineTsdbStore implements TsdbStore {
    public Map<SeriesKey, List<BucketAggregate>> bucketedAggregate(SeriesFilter filter, AggregateFunction fn,
                                                                   TimeWindow window, Duration bucketWidth,
                                                                   Double percentile, TsdbDeadline deadline) {
+        // TDengine PERCENTILE refuses supertable queries (including INTERVAL
+        // windows); per-series subtable scans keep it exact on the raw path —
+        // this store declares rollupSupport=NONE, so tier reads never apply.
+        if (fn == AggregateFunction.PERCENTILE) {
+            Map<SeriesKey, List<BucketAggregate>> result = new LinkedHashMap<>();
+            if (filter.tenantWide()) {
+                throw new IllegalArgumentException(
+                        "TDengine percentile needs explicit series; tenant-wide percentile is unsupported");
+            }
+            for (SeriesKey key : filter.series()) {
+                String sql = "SELECT CAST(_wstart AS BIGINT) AS bucket, %s AS agg_value, COUNT(*) AS sample_count FROM %s WHERE ts >= %d AND ts < %d INTERVAL(%d)"
+                        .formatted(aggregateExpression(fn, percentile), subtable(key),
+                                microsOf(window.from()), microsOf(window.toExclusive()),
+                                bucketWidth.toNanos() / 1000);
+                timedVoid(deadline, () -> {
+                    List<Map<String, Object>> rows = jdbc.queryForList(sql);
+                    List<BucketAggregate> buckets = new ArrayList<>();
+                    for (Map<String, Object> row : rows) {
+                        buckets.add(new BucketAggregate(instantOfMicros(((Number) row.get("bucket")).longValue()),
+                                Objects.nonNull(row.get("agg_value")) ? ((Number) row.get("agg_value")).doubleValue() : null,
+                                ((Number) row.get("sample_count")).longValue()));
+                    }
+                    if (!buckets.isEmpty()) {
+                        result.put(key, buckets);
+                    }
+                });
+            }
+            return result;
+        }
        String sql = """
                SELECT tenant_id, device_id, point_id, CAST(_wstart AS BIGINT) AS bucket,
                       %s AS agg_value, COUNT(*) AS sample_count
@@ -85,10 +85,26 @@ public final class TimescaleTsdbStore implements TsdbStore {

    private static final int SERIES_IN_CHUNK = 500;

+    /**
+     * S16 rollup tiers. These are the observability pipeline's continuous
+     * aggregates (07-iot-dc3-observability.sql, queried by Grafana) — one
+     * structure serves both consumers; the adapter recreates them with an
+     * identical shape when it boots standalone.
+     */
+    private static final String ROLLUP_1M = "cagg_point_value_1m";
+
+    private static final String ROLLUP_1H = "cagg_point_value_1h";
+
+    private static final java.time.Duration TIER_MINUTE = java.time.Duration.ofMinutes(1);
+
+    private static final java.time.Duration TIER_HOUR = java.time.Duration.ofHours(1);
+
    private static final String COLUMNS = "tenant_id, device_id, point_id, message_id, schema_version, "
            + "driver_node, sequence, fencing_token, raw_value, cal_value, num_value, quality, "
            + "driver_id, create_time, operate_time";

+    private final int minuteTierKeepDays;
+
    private final JdbcTemplate jdbc;

    private static final RowMapper<PointValueSample> SAMPLE_MAPPER = (rs, i) -> new PointValueSample(
@@ -102,6 +118,11 @@ public final class TimescaleTsdbStore implements TsdbStore {
            rs.getLong("fencing_token"), rs.getLong("driver_id"));

    public TimescaleTsdbStore(DataSource dataSource) {
+        this(dataSource, 365);
+    }
+
+    public TimescaleTsdbStore(DataSource dataSource, int minuteTierKeepDays) {
+        this.minuteTierKeepDays = minuteTierKeepDays;
        this.jdbc = new JdbcTemplate(dataSource);
        bootstrap();
    }
@@ -151,9 +172,72 @@ public final class TimescaleTsdbStore implements TsdbStore {
                CREATE INDEX IF NOT EXISTS idx_point_value_tenant_time
                ON %s (tenant_id, create_time DESC)""".formatted(TABLE));
        primeInitialChunk();
+        bootstrapRollups();
        log.info("Timescale store ready (table {})", TABLE);
    }

+    /**
+     * S16 tiered lifecycle: raw → 1-minute rollup → 1-hour rollup. Both tiers are
+     * real-time continuous aggregates ({@code materialized_only = FALSE}) so reads
+     * are correct immediately after an append — the background refresh policies
+     * only move the aggregation work off the read path. Aggregates are chosen for
+     * exact composition across tiers: AVG recombines as SUM(sum)/SUM(numeric_count),
+     * FIRST/LAST re-select by bucket time; PERCENTILE stays on the raw path.
+     * The hour tier groups by the explicit time_bucket expression because its
+     * output alias collides with the source column name.
+     */
+    private void bootstrapRollups() {
+        try {
+            jdbc.execute("""
+                    CREATE MATERIALIZED VIEW IF NOT EXISTS %s
+                    WITH (timescaledb.continuous, timescaledb.materialized_only = FALSE) AS
+                    SELECT time_bucket(INTERVAL '1 minute', create_time) AS bucket,
+                           tenant_id, driver_id, device_id, point_id,
+                           COUNT(*) AS sample_count, COUNT(num_value) AS num_count,
+                           AVG(num_value) AS num_avg, MIN(num_value) AS num_min,
+                           MAX(num_value) AS num_max, SUM(num_value) AS num_sum,
+                           FIRST(cal_value, create_time) AS cal_first,
+                           LAST(cal_value, create_time) AS cal_last
+                    FROM %s
+                    GROUP BY time_bucket(INTERVAL '1 minute', create_time),
+                             tenant_id, driver_id, device_id, point_id
+                    WITH NO DATA""".formatted(ROLLUP_1M, TABLE));
+            jdbc.execute("CREATE INDEX IF NOT EXISTS idx_cagg_pv_1m_lookup ON %s (tenant_id, device_id, point_id, bucket DESC)".formatted(ROLLUP_1M));
+            jdbc.execute("""
+                    CREATE MATERIALIZED VIEW IF NOT EXISTS %s
+                    WITH (timescaledb.continuous, timescaledb.materialized_only = FALSE) AS
+                    SELECT time_bucket(INTERVAL '1 hour', bucket) AS bucket,
+                           tenant_id, driver_id, device_id, point_id,
+                           SUM(sample_count) AS sample_count, SUM(num_count) AS num_count,
+                           SUM(num_sum) / NULLIF(SUM(num_count), 0) AS num_avg,
+                           MIN(num_min) AS num_min, MAX(num_max) AS num_max,
+                           SUM(num_sum) AS num_sum,
+                           FIRST(cal_first, bucket) AS cal_first,
+                           LAST(cal_last, bucket) AS cal_last
+                    FROM %s
+                    GROUP BY time_bucket(INTERVAL '1 hour', bucket),
+                             tenant_id, driver_id, device_id, point_id
+                    WITH NO DATA""".formatted(ROLLUP_1H, ROLLUP_1M));
+            jdbc.execute("CREATE INDEX IF NOT EXISTS idx_cagg_pv_1h_lookup ON %s (tenant_id, device_id, point_id, bucket DESC)".formatted(ROLLUP_1H));
+            jdbc.execute("""
+                    SELECT add_continuous_aggregate_policy('%s', if_not_exists => TRUE,
+                            start_offset => NULL, end_offset => INTERVAL '1 minute',
+                            schedule_interval => INTERVAL '1 minute')""".formatted(ROLLUP_1M));
+            jdbc.execute("""
+                    SELECT add_continuous_aggregate_policy('%s', if_not_exists => TRUE,
+                            start_offset => NULL, end_offset => INTERVAL '5 minutes',
+                            schedule_interval => INTERVAL '5 minutes')""".formatted(ROLLUP_1H));
+            jdbc.execute("""
+                    SELECT add_retention_policy('%s', INTERVAL '%d days', if_not_exists => TRUE)"""
+                    .formatted(ROLLUP_1M, minuteTierKeepDays));
+            // The hour tier is kept forever — no retention policy.
+        } catch (DataAccessException e) {
+            // Plain-PG deployments (no timescaledb extension) skip the tiers; reads
+            // stay correct on the raw path.
+            log.warn("Rollup tiers not created, reads stay on the raw path: {}", e.getMessage());
+        }
+    }
+
    /**
     * TimescaleDB sizes the initial chunk around the first inserted row; multi-row or
     * microsecond-boundary first appends can leave that row outside the chunk's final
@@ -185,7 +269,7 @@ public final class TimescaleTsdbStore implements TsdbStore {
    public TsdbCapabilities capabilities() {
        return new TsdbCapabilities(
                true, true, true, true, true,
-                RollupSupport.NONE, 5000,
+                RollupSupport.NATIVE, 5000,
                true, OrderingGuarantee.PER_SERIES, Precision.MICRO, true, true);
    }

@@ -359,6 +443,10 @@ public final class TimescaleTsdbStore implements TsdbStore {
    public Map<SeriesKey, List<BucketAggregate>> bucketedAggregate(SeriesFilter filter, AggregateFunction fn,
                                                                   TimeWindow window, Duration bucketWidth,
                                                                   Double percentile, TsdbDeadline deadline) {
+        String tier = rollupTierFor(fn, bucketWidth);
+        if (Objects.nonNull(tier)) {
+            return bucketedAggregateFromTier(tier, filter, fn, window, bucketWidth, deadline);
+        }
        String expr = aggregateExpression(fn, percentile);
        String sql = """
                SELECT tenant_id, device_id, point_id, time_bucket(?::interval, create_time) AS bucket,
@@ -403,6 +491,21 @@ public final class TimescaleTsdbStore implements TsdbStore {
    @Override
    public List<BucketAggregate> bucketedCount(long tenantId, TimeWindow window,
                                               Duration bucketWidth, TsdbDeadline deadline) {
+        String tier = rollupTierFor(AggregateFunction.COUNT, bucketWidth);
+        if (Objects.nonNull(tier)) {
+            // Tier reads cover whole tier granules inside the window: buckets are
+            // precomputed counts summed per coarse bucket.
+            String sql = """
+                    SELECT time_bucket(?::interval, bucket) AS bucket, SUM(sample_count) AS sample_count
+                    FROM %s WHERE tenant_id = ? AND bucket >= ? AND bucket < ?
+                    GROUP BY 1 ORDER BY 1 ASC""".formatted(tier);
+            Object[] args = {bucketWidth.toMillis() + " milliseconds", tenantId,
+                    OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC),
+                    OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC)};
+            return timed(deadline, () -> jdbc.query(sql, (rs, i) -> new BucketAggregate(
+                    toInstant(rs, 1).truncatedTo(java.time.temporal.ChronoUnit.MILLIS),
+                    null, rs.getLong(2)), args));
+        }
        String sql = """
                SELECT time_bucket(?::interval, create_time) AS bucket, COUNT(*) AS sample_count
                FROM %s WHERE tenant_id = ? AND create_time >= ? AND create_time < ?
@@ -521,6 +624,25 @@ public final class TimescaleTsdbStore implements TsdbStore {
                series.tenantId(), series.deviceId(), series.pointId(),
                OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC),
                OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC));
+        refreshTiers(window);
+    }
+
+    private void refreshTiers(TimeWindow window) {
+        // The real-time tier reads reflect the delete immediately; re-materialize
+        // so the stored tier rows converge too (tenant offboarding is rare).
+        try {
+            for (String tier : new String[]{ROLLUP_1M, ROLLUP_1H}) {
+                jdbc.query("CALL refresh_continuous_aggregate(?, ?, ?)",
+                        ps -> {
+                            ps.setString(1, tier);
+                            ps.setObject(2, OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC));
+                            ps.setObject(3, OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC));
+                        }, rs -> null);
+            }
+        } catch (DataAccessException e) {
+            log.warn("Rollup refresh after deleteRange failed; background policies will converge: {}",
+                    e.getMessage());
+        }
    }

    @Override
@@ -549,6 +671,71 @@ public final class TimescaleTsdbStore implements TsdbStore {

    // ===== helpers =====

+    /**
+     * S16 read transparency: bucket widths at or above a materialized tier's
+     * granularity are served from that tier (PERCENTILE never — it is not
+     * materialized). Tier reads cover whole tier granules inside the window.
+     */
+    private String rollupTierFor(AggregateFunction fn, Duration bucketWidth) {
+        // FIRST/LAST stay raw: the shared observability caggs carry first/last of
+        // the textual cal_value only, and serving numeric first/last from tiers
+        // would version-skew deployments created before that shape existed.
+        if (fn == AggregateFunction.PERCENTILE || fn == AggregateFunction.FIRST
+                || fn == AggregateFunction.LAST) {
+            return null;
+        }
+        if (!bucketWidth.minus(TIER_HOUR).isNegative()) {
+            return ROLLUP_1H;
+        }
+        if (!bucketWidth.minus(TIER_MINUTE).isNegative()) {
+            return ROLLUP_1M;
+        }
+        return null;
+    }
+
+    private Map<SeriesKey, List<BucketAggregate>> bucketedAggregateFromTier(String tier, SeriesFilter filter,
+                                                                            AggregateFunction fn, TimeWindow window,
+                                                                            Duration bucketWidth, TsdbDeadline deadline) {
+        String sql = """
+                SELECT time_bucket(?::interval, bucket) AS bucket, tenant_id, device_id, point_id,
+                       %s AS value, SUM(sample_count) AS sample_count
+                FROM %s v WHERE %s AND v.bucket >= ? AND v.bucket < ?
+                GROUP BY 1, 2, 3, 4
+                ORDER BY 1 ASC"""
+                .formatted(tierExpression(fn), tier, seriesWhere(filter));
+        List<Object> args = new ArrayList<>();
+        args.add(bucketWidth.toMillis() + " milliseconds");
+        args.addAll(seriesArgs(filter));
+        args.add(OffsetDateTime.ofInstant(window.from(), ZoneOffset.UTC));
+        args.add(OffsetDateTime.ofInstant(window.toExclusive(), ZoneOffset.UTC));
+        Map<SeriesKey, List<BucketAggregate>> result = new LinkedHashMap<>();
+        timedVoid(deadline, () -> {
+            List<Map<String, Object>> rows = jdbc.queryForList(sql, args.toArray());
+            for (Map<String, Object> row : rows) {
+                result.computeIfAbsent(new SeriesKey(((Number) row.get("tenant_id")).longValue(),
+                                ((Number) row.get("device_id")).longValue(),
+                                ((Number) row.get("point_id")).longValue()), k -> new ArrayList<>())
+                        .add(new BucketAggregate(((java.sql.Timestamp) row.get("bucket")).toInstant()
+                                .truncatedTo(java.time.temporal.ChronoUnit.MILLIS),
+                                (Double) row.get("value"),
+                                ((Number) row.get("sample_count")).longValue()));
+            }
+        });
+        return result;
+    }
+
+    /** Exactly composable re-aggregation over the shared observability tiers. */
+    private static String tierExpression(AggregateFunction fn) {
+        return switch (fn) {
+            case AVG -> "SUM(num_sum) / NULLIF(SUM(num_count), 0)";
+            case MIN -> "MIN(num_min)";
+            case MAX -> "MAX(num_max)";
+            case SUM -> "SUM(num_sum)";
+            case COUNT -> "CAST(SUM(sample_count) AS DOUBLE PRECISION)";
+            case FIRST, LAST, PERCENTILE -> throw new IllegalArgumentException("never tiered: " + fn);
+        };
+    }
+
    private static Instant toInstant(ResultSet rs, String column) throws SQLException {
        Timestamp ts = rs.getTimestamp(column);
        return Objects.isNull(ts) ? null : ts.toInstant();
@@ -26,6 +26,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;

 import javax.sql.DataSource;
@@ -51,8 +52,9 @@ public class TsdbTimescaleAutoConfiguration {

    @Bean
    @ConditionalOnMissingBean(TsdbStore.class)
-    public TsdbStore tsdbStore(@Qualifier("tsdbDataSource") DataSource tsdbDataSource) {
-        TsdbStore store = new TimescaleTsdbStore(tsdbDataSource);
+    public TsdbStore tsdbStore(@Qualifier("tsdbDataSource") DataSource tsdbDataSource,
+                               @Value("${dc3.tsdb.timescale.rollup.minute-keep-days:365}") int minuteTierKeepDays) {
+        TsdbStore store = new TimescaleTsdbStore(tsdbDataSource, minuteTierKeepDays);
        log.info("TSDB port negotiated, store={}, capabilities={}", store.type(), store.capabilities());
        return store;
    }
@@ -99,7 +99,10 @@ ALTER TABLE dc3_point_value
        timescaledb.compress_orderby = 'create_time DESC'
    );
 SELECT public.add_compression_policy('dc3_point_value', INTERVAL '7 days');
-SELECT public.add_retention_policy('dc3_point_value', INTERVAL '180 days');
+-- S16 tiered lifecycle (docs/design/tsdb-abstraction.md §9.6): raw samples are
+-- kept 30 days; the 1-minute rollup (1 year) and 1-hour rollup (forever) are
+-- real-time continuous aggregates created by the timescale adapter bootstrap.
+SELECT public.add_retention_policy('dc3_point_value', INTERVAL '30 days');

 -- ----------------------------
 -- Transactional latest-value projection
@@ -114,8 +114,9 @@ COMMENT ON VIEW cagg_point_value_1h IS 'One-hour hierarchical point-value contin
 -- ----------------------------
 -- Background refresh of the materialized window. Realtime mode still
 -- answers queries from the live tail; the policy only backfills durable
-- buckets. start_offset must stay within the hypertable retention window
-- (180 days, see 05-iot-dc3-history.sql).
+-- buckets. Raw samples are kept 30 days (05-iot-dc3-history.sql, S16 tiered
+-- lifecycle); the refresh offsets predate that on purpose so rebuilding the
+-- materialized tiers after a drop still folds whatever raw history exists.
 SELECT add_continuous_aggregate_policy('cagg_point_value_1m',
                                       start_offset => INTERVAL '7 days',
                                       end_offset => INTERVAL '1 minute',
@@ -126,6 +127,11 @@ SELECT add_continuous_aggregate_policy('cagg_point_value_1h',
                                       end_offset => INTERVAL '5 minutes',
                                       schedule_interval => INTERVAL '5 minutes');

+-- S16 tiered lifecycle: the minute tier ages out after one year; the hour tier
+-- is kept forever. The timescale adapter registers the same policy with
+-- if_not_exists, so embedded and standalone deployments converge.
+SELECT add_retention_policy('cagg_point_value_1m', INTERVAL '365 days');
+
 -- ----------------------------
 -- Device metadata view for Grafana
 -- ----------------------------
@@ -411,6 +411,35 @@ REST 控制器"工作（目录从 OpenAPI 快照合成），因此"九工具"的
 - `DataAnnotationGateTest` 守门要求每个请求字段带 `@Schema` 描述——正是
  inputSchema 质量的机械化保障，本次全部满足。

+**Phase 2 第三片（2026-08-21，S16 多级保留 + rollup 首发验证，Phase 2 至此完成）**：
+生命周期落地为 **raw 30 天（seed 05）→ 1 分钟层 1 年 → 1 小时层永久**（`dc3.tsdb.
+timescale.rollup.minute-keep-days` 可配）。实施中的关键决策与教训：
+1. **与 observability 管线合并而非并存**：seed 07 早已为 Grafana 建了
+   `cagg_point_value_1m/1h`（real-time、含 driver 维度与 cal_first/cal_last）——
+   适配器最初另建一对 cagg 会造成双份物化开销。收敛为**单一结构**：适配器
+   以 `IF NOT EXISTS` + 与 seed 完全一致的列集引导同名 cagg，嵌入式/独立部署
+   两条路径汇合；Grafana 与 port 读同一份物化。
+2. **FIRST/LAST 留在原始路径**：共享 cagg 只有 cal_value 的首末（文本），数值
+   首末列在既有部署的 cagg 里不存在——从层上供 FIRST/LAST 会造成新旧部署
+   行为分叉；M4 典型窗口（≤30 天）本就落在原始保留期内，原始扫描即可。
+3. **real-time cagg 是"读即刻正确"的关键**：TS 2.13+ 新建 cagg 默认
+   `materialized_only=TRUE`（物化前读到 0 行），显式 `materialized_only=FALSE`
+   后 TCK 在**未等任何刷新**的情况下断言分级 COUNT/AVG 与原始扫描逐位一致；
+   刷新策略只把聚合工作挪去后台。
+4. **AVG 从层重组必须 SUM(sum)/SUM(count)**（对每桶平均值再平均是错的）；
+   COUNT=SUM(sample_count)。cagg 列集为此带 num_sum/num_count。
+5. SQL 教训两则：层级 cagg 的 `GROUP BY` 不能用与源列同名的别名（绑到源列，
+   seed 07 早已注释过同款坑——GROUP BY 显式表达式或序数）；带参数的
+   time_bucket 在 SELECT 与 GROUP BY 里是两个不同表达式，**GROUP BY 用序数**
+   才引用同一输出列。
+6. TDengine 案例二十暴露 `bucketedAggregate PERCENTILE` 超级表限制——补了
+   逐序列子表分支（与 aggregate() 同款策略），TCK 断言分桶百分位不再报
+   "percentile is only supported in single table query"。
+门：timescale 24/24（含新案例"分级读与原始扫描一致"：COUNT/AVG/LAST/
+bucketedCount/分桶 P50 五路验证）+ TDengine 24/24（2 例按声明能力跳过，
+NONE 档诚实原始降级同样全过）。deleteRange 后对两层补
+refresh_continuous_aggregate，real-time 部分即刻正确、物化部分随后收敛。
+
 ## 7. 逐库映射

 | 概念 | TimescaleDB | TDengine 3.x | InfluxDB 3 | IoTDB |