You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by go...@apache.org on 2026/09/03 09:11:36 UTC
(pinot) branch cbo-1-stats-contracts-and-stores created (now b8878b58f94)
This is an automated email from the ASF dual-hosted git repository.
gortiz pushed a change to branch cbo-1-stats-contracts-and-stores
in repository https://gitbox.apache.org/repos/asf/pinot.git
at b8878b58f94 Add SQLite and in-memory statistics stores
This branch includes the following new commits:
new 3a1b835cc59 Add statistics contracts for cost-based optimization
new b8878b58f94 Add SQLite and in-memory statistics stores
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
(pinot) 02/02: Add SQLite and in-memory statistics stores
Posted by go...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gortiz pushed a commit to branch cbo-1-stats-contracts-and-stores
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit b8878b58f943f7dce73f10676051d69155da636c
Author: Gonzalo Ortiz <19...@users.noreply.github.com>
AuthorDate: Thu Aug 27 20:02:17 2026 +0200
Add SQLite and in-memory statistics stores
Statistics for hundreds of thousands of segments must not compete with query
execution for JVM heap, so the default store is an embedded SQLite database
(WAL, a small read-connection pool, and drop-and-rebuild recovery if the file is
ever unreadable). Surviving a restart is a side benefit rather than the goal:
the file also lets the broker skip re-collecting segments whose crc has not
changed.
There is no migration framework. SQLite's own user_version pragma carries the
schema version, and a store whose version does not match is discarded and
rebuilt rather than migrated. Every row here is derived from ZooKeeper metadata
the broker re-reads at startup, so a rebuild costs nothing that was not already
being read -- and it puts schema change, corruption and an unreadable file on
one recovery path. This adds exactly one third-party dependency, sqlite-jdbc.
InMemoryStatsStore is the alternative for brokers that cannot or should not
write a file. It keeps the same semantics -- consuming segments excluded from
aggregates but present for crc reconciliation, no-data reported as absent rather
than zero -- by folding through the same StatsAggregations helpers.
Both serve table-level reads from a rollup recomputed only after a write, since
query planning asks for them on every compile while writes arrive at segment
push cadence. The version stamp is what makes that safe without holding the
write lock: a rollup computed from rows that changed underneath it is used for
that call but never published.
Tests are written as a contract both stores must satisfy -- the optimizer must
not change with the configured store -- with each implementation adding only
what is specific to it: durability and corruption recovery for SQLite, starting
empty for in-memory.
---
LICENSE-binary | 4 +
pinot-broker/pom.xml | 8 +
.../pinot/broker/stats/InMemoryStatsStore.java | 357 +++++++++
.../pinot/broker/stats/SqliteStatsStore.java | 885 +++++++++++++++++++++
.../pinot/broker/stats/InMemoryStatsStoreTest.java | 52 ++
.../pinot/broker/stats/SqliteStatsStoreTest.java | 259 ++++++
.../pinot/broker/stats/StatsStoreContractTest.java | 560 +++++++++++++
pom.xml | 8 +
8 files changed, 2133 insertions(+)
diff --git a/LICENSE-binary b/LICENSE-binary
index 30034937325..956d81c0d6f 100644
--- a/LICENSE-binary
+++ b/LICENSE-binary
@@ -542,6 +542,7 @@ org.scala-lang:scala-reflect:2.13.18
org.slf4j:jcl-over-slf4j:2.0.17
org.webjars:swagger-ui:5.32.0
org.xerial.snappy:snappy-java:1.1.10.8
+org.xerial:sqlite-jdbc:3.46.1.3
org.yaml:snakeyaml:2.6
software.amazon.awssdk:annotations:2.42.11
software.amazon.awssdk:apache-client:2.42.11
@@ -627,6 +628,9 @@ BSD 2-Clause
------------
com.github.luben:zstd-jni:1.5.7-7
org.codehaus.woodstox:stax2-api:4.2.2
+org.xerial:sqlite-jdbc:3.46.1.3 (the JDBC layer derived from Zentus SQLiteJDBC;
+ the remainder of the artifact is Apache-2.0, listed above)
+ Copyright (c) 2006, David Crawshaw. All rights reserved.
BSD 3-Clause
diff --git a/pinot-broker/pom.xml b/pinot-broker/pom.xml
index b96db348c40..ba90112a91b 100644
--- a/pinot-broker/pom.xml
+++ b/pinot-broker/pom.xml
@@ -33,6 +33,10 @@
<pinot.root>${basedir}/..</pinot.root>
</properties>
<dependencies>
+ <dependency>
+ <groupId>org.apache.pinot</groupId>
+ <artifactId>pinot-query-planner-spi</artifactId>
+ </dependency>
<dependency>
<groupId>org.apache.pinot</groupId>
<artifactId>pinot-materialized-view</artifactId>
@@ -53,6 +57,10 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.xerial</groupId>
+ <artifactId>sqlite-jdbc</artifactId>
+ </dependency>
<!-- Test -->
<dependency>
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/stats/InMemoryStatsStore.java b/pinot-broker/src/main/java/org/apache/pinot/broker/stats/InMemoryStatsStore.java
new file mode 100644
index 00000000000..22f7ee8ef11
--- /dev/null
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/stats/InMemoryStatsStore.java
@@ -0,0 +1,357 @@
+/**
+ * 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.pinot.broker.stats;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.annotation.Nullable;
+import org.apache.pinot.query.planner.spi.stats.ColumnStatistics;
+import org.apache.pinot.query.planner.spi.stats.SegmentColumnStatsRow;
+import org.apache.pinot.query.planner.spi.stats.SegmentStatsRow;
+import org.apache.pinot.query.planner.spi.stats.StatConfidence;
+import org.apache.pinot.query.planner.spi.stats.StatsAggregations;
+import org.apache.pinot.query.planner.spi.stats.StatsStore;
+import org.apache.pinot.query.planner.spi.stats.StatsStoreException;
+import org.apache.pinot.query.planner.spi.stats.TableStatistics;
+
+
+/// Heap-resident [StatsStore], for deployments that do not want a database file on the broker
+/// (read-only filesystems, ephemeral containers, tests).
+///
+/// ### What it trades away
+/// Statistics do not survive a restart. For the base tier that costs little: the values are derived
+/// from the ZooKeeper segment metadata the broker re-reads at startup anyway, so an empty store is
+/// simply re-populated by the same listener callbacks — [#getSegmentCrcs] returns an empty map and
+/// every segment is treated as new. Per-column statistics pulled from servers are the expensive
+/// case, since those must be fetched again.
+///
+/// Table-level reads (row counts, consuming-segment detection) are served from a rollup that is
+/// recomputed only after a write, so query planning does not scan the segment map. Time-range
+/// estimates still scan the table's committed segments, since the answer depends on the requested
+/// range.
+///
+/// Heap use is proportional to the number of segments (and stored columns) of the tables this
+/// broker serves. The SQLite-backed store exists precisely to keep that off-heap, so prefer this
+/// implementation only when the segment count is modest or a file is unacceptable.
+///
+/// ### Semantics
+/// Identical to the SQLite-backed store, and deliberately expressed through the same
+/// [StatsAggregations] helpers so the two cannot drift: consuming segments are excluded from
+/// table-level, column-level and time-range reads but included by [#getSegmentCrcs]; a column row
+/// only counts while its segment row exists; "no data" is reported as `null` /
+/// [OptionalLong#empty()] and never as a zero estimate.
+///
+/// ### Thread-safety
+/// Safe for concurrent reads with a single concurrent writer, as the interface requires. Reads
+/// aggregate over concurrent maps and therefore observe a weakly-consistent view: a read racing a
+/// multi-row upsert may see part of that batch. Unlike a SQLite transaction the batch is not
+/// atomic, which is acceptable because every consumer treats these values as estimates.
+public class InMemoryStatsStore implements StatsStore {
+
+ /// table name → segment name → row.
+ private final Map<String, Map<String, StoredSegment>> _segments = new ConcurrentHashMap<>();
+ /// table name → segment name → column name → row.
+ private final Map<String, Map<String, Map<String, SegmentColumnStatsRow>>> _columns = new ConcurrentHashMap<>();
+ /// table name → monotonically increasing write counter, bumped after every mutation of that table.
+ private final Map<String, AtomicLong> _versions = new ConcurrentHashMap<>();
+ /// table name → table-level rollup, valid only while its version matches the table's counter.
+ private final Map<String, CachedAggregate> _aggregates = new ConcurrentHashMap<>();
+
+ private volatile boolean _closed;
+
+ @Override
+ public void init() {
+ // Nothing to open: the maps are ready on construction.
+ }
+
+ @Override
+ public void upsertSegmentStats(String tableNameWithType, List<SegmentStatsRow> rows)
+ throws StatsStoreException {
+ checkOpen();
+ long now = System.currentTimeMillis();
+ Map<String, StoredSegment> table = _segments.computeIfAbsent(tableNameWithType, k -> new ConcurrentHashMap<>());
+ for (SegmentStatsRow row : rows) {
+ table.put(row.segmentName(), new StoredSegment(row, now));
+ }
+ invalidate(tableNameWithType);
+ }
+
+ @Override
+ public void upsertSegmentColumnStats(String tableNameWithType, List<SegmentColumnStatsRow> rows)
+ throws StatsStoreException {
+ checkOpen();
+ Map<String, Map<String, SegmentColumnStatsRow>> table =
+ _columns.computeIfAbsent(tableNameWithType, k -> new ConcurrentHashMap<>());
+ for (SegmentColumnStatsRow row : rows) {
+ table.computeIfAbsent(row.segmentName(), k -> new ConcurrentHashMap<>()).put(row.columnName(), row);
+ }
+ }
+
+ @Override
+ public void removeSegments(String tableNameWithType, Collection<String> segmentNames)
+ throws StatsStoreException {
+ checkOpen();
+ Map<String, StoredSegment> segments = _segments.get(tableNameWithType);
+ Map<String, Map<String, SegmentColumnStatsRow>> columns = _columns.get(tableNameWithType);
+ for (String segmentName : segmentNames) {
+ if (segments != null) {
+ segments.remove(segmentName);
+ }
+ if (columns != null) {
+ columns.remove(segmentName);
+ }
+ }
+ // Drop emptied tables, so getTables() means "holds statistics" here exactly as it does in a
+ // row-backed store -- it is the input to a destructive purge, so the two must not disagree.
+ if (segments != null && segments.isEmpty()) {
+ _segments.remove(tableNameWithType, segments);
+ }
+ if (columns != null && columns.isEmpty()) {
+ _columns.remove(tableNameWithType, columns);
+ }
+ if (_segments.get(tableNameWithType) == null && _columns.get(tableNameWithType) == null) {
+ // Nothing left to roll up; see purgeTable for why this removes instead of invalidating.
+ _aggregates.remove(tableNameWithType);
+ _versions.remove(tableNameWithType);
+ } else {
+ invalidate(tableNameWithType);
+ }
+ }
+
+ @Override
+ public Map<String, Long> getSegmentCrcs(String tableNameWithType)
+ throws StatsStoreException {
+ checkOpen();
+ Map<String, StoredSegment> segments = _segments.get(tableNameWithType);
+ // Mutable and detached, like the SQLite store's result: the SPI does not promise immutability
+ // and callers may adjust the map.
+ if (segments == null) {
+ return new HashMap<>();
+ }
+ // Includes consuming segments: reconciliation must not re-upsert them on every restart.
+ Map<String, Long> crcs = new HashMap<>(segments.size());
+ for (Map.Entry<String, StoredSegment> entry : segments.entrySet()) {
+ crcs.put(entry.getKey(), entry.getValue()._row.crc());
+ }
+ return crcs;
+ }
+
+ @Override
+ @Nullable
+ public TableStatistics getTableStats(String tableNameWithType)
+ throws StatsStoreException {
+ checkOpen();
+ return aggregate(tableNameWithType)._stats;
+ }
+
+ @Override
+ @Nullable
+ public ColumnStatistics getColumnStats(String tableNameWithType, String columnName)
+ throws StatsStoreException {
+ checkOpen();
+ Map<String, Map<String, SegmentColumnStatsRow>> columnsBySegment = _columns.get(tableNameWithType);
+ Map<String, StoredSegment> segments = _segments.get(tableNameWithType);
+ if (columnsBySegment == null || segments == null) {
+ return null;
+ }
+
+ // Same fold as the SQLite store, through the shared accumulator, so the two cannot disagree.
+ StatsAggregations.ColumnStatsAccumulator accumulator = new StatsAggregations.ColumnStatsAccumulator();
+ for (Map.Entry<String, Map<String, SegmentColumnStatsRow>> entry : columnsBySegment.entrySet()) {
+ SegmentColumnStatsRow column = entry.getValue().get(columnName);
+ if (column == null) {
+ continue;
+ }
+ // Mirrors the SQL join onto segment_stats: a column row without a live, committed segment row
+ // contributes nothing, and the segment row supplies the doc count used for weighting.
+ StoredSegment segment = segments.get(entry.getKey());
+ if (segment == null || segment._row.consuming()) {
+ continue;
+ }
+ accumulator.add(segment._row.totalDocs(), column);
+ }
+ return accumulator.isEmpty() ? null : accumulator.build(columnName);
+ }
+
+ @Override
+ public OptionalLong estimateRowsInTimeRange(String tableNameWithType, long startMs, long endMs)
+ throws StatsStoreException {
+ checkOpen();
+ Map<String, StoredSegment> segments = _segments.get(tableNameWithType);
+ if (segments == null) {
+ return OptionalLong.empty();
+ }
+ long totalRows = 0;
+ boolean hasCommittedSegment = false;
+ for (StoredSegment segment : segments.values()) {
+ SegmentStatsRow row = segment._row;
+ if (row.consuming()) {
+ continue;
+ }
+ // Distinguishes "no overlapping segment" (a real estimate of 0) from "no statistics"
+ // (empty), exactly as the SQL existence sentinel does.
+ hasCommittedSegment = true;
+ totalRows += StatsAggregations.overlapRows(row.totalDocs(), row.startTimeMs(), row.endTimeMs(),
+ startMs, endMs);
+ }
+ return hasCommittedSegment ? OptionalLong.of(totalRows) : OptionalLong.empty();
+ }
+
+ @Override
+ public Set<String> getTables()
+ throws StatsStoreException {
+ checkOpen();
+ Set<String> tables = new HashSet<>(_segments.keySet());
+ tables.addAll(_columns.keySet());
+ return tables;
+ }
+
+ @Override
+ public void purgeTable(String tableNameWithType)
+ throws StatsStoreException {
+ checkOpen();
+ _segments.remove(tableNameWithType);
+ _columns.remove(tableNameWithType);
+ // Remove rather than invalidate: invalidate() would re-create the version entry it just
+ // dropped, so a broker that serves many tables over its lifetime would accumulate one entry
+ // per table it ever saw.
+ _aggregates.remove(tableNameWithType);
+ _versions.remove(tableNameWithType);
+ }
+
+ @Override
+ public void purgeAll()
+ throws StatsStoreException {
+ checkOpen();
+ _segments.clear();
+ _columns.clear();
+ _aggregates.clear();
+ _versions.clear();
+ }
+
+ @Override
+ public boolean hasConsumingSegments(String tableNameWithType)
+ throws StatsStoreException {
+ checkOpen();
+ return aggregate(tableNameWithType)._hasConsuming;
+ }
+
+ @Override
+ public void close() {
+ _closed = true;
+ _segments.clear();
+ _columns.clear();
+ _aggregates.clear();
+ _versions.clear();
+ }
+
+ /// Returns the table-level rollup, recomputing it only when the table changed since it was last
+ /// computed. Query planning asks for these on every compile, so an O(#segments) scan per call
+ /// would show up directly in planning latency; writes are comparatively rare.
+ ///
+ /// The version is read before the scan and re-checked after it, so a rollup computed from a state
+ /// a concurrent writer has already replaced is used once but never cached.
+ private CachedAggregate aggregate(String tableNameWithType) {
+ AtomicLong version = _versions.computeIfAbsent(tableNameWithType, k -> new AtomicLong());
+ long observed = version.get();
+ CachedAggregate cached = _aggregates.get(tableNameWithType);
+ if (cached != null && cached._version == observed) {
+ return cached;
+ }
+ CachedAggregate computed = computeAggregate(tableNameWithType, observed);
+ if (version.get() == observed) {
+ _aggregates.put(tableNameWithType, computed);
+ }
+ return computed;
+ }
+
+ private CachedAggregate computeAggregate(String tableNameWithType, long version) {
+ Map<String, StoredSegment> segments = _segments.get(tableNameWithType);
+ if (segments == null) {
+ return new CachedAggregate(version, null, false);
+ }
+ long totalDocs = 0;
+ long sizeBytes = 0;
+ long maxUpdatedAt = 0;
+ int committed = 0;
+ boolean hasConsuming = false;
+ for (StoredSegment segment : segments.values()) {
+ if (segment._row.consuming()) {
+ hasConsuming = true;
+ continue;
+ }
+ totalDocs += segment._row.totalDocs();
+ sizeBytes += segment._row.sizeBytes();
+ maxUpdatedAt = Math.max(maxUpdatedAt, segment._updatedAtMs);
+ committed++;
+ }
+ if (committed == 0) {
+ return new CachedAggregate(version, null, hasConsuming);
+ }
+ return new CachedAggregate(version, TableStatistics.builder()
+ .rowCount(totalDocs, StatConfidence.EXACT)
+ .tableSizeBytes(sizeBytes, StatConfidence.EXACT)
+ .updatedAtMs(maxUpdatedAt)
+ .build(), hasConsuming);
+ }
+
+ private void invalidate(String tableNameWithType) {
+ _versions.computeIfAbsent(tableNameWithType, k -> new AtomicLong()).incrementAndGet();
+ }
+
+ private void checkOpen()
+ throws StatsStoreException {
+ if (_closed) {
+ throw new StatsStoreException("InMemoryStatsStore is closed");
+ }
+ }
+
+ /// A table-level rollup together with the write version it was computed from.
+ private static final class CachedAggregate {
+ private final long _version;
+ @Nullable
+ private final TableStatistics _stats;
+ private final boolean _hasConsuming;
+
+ CachedAggregate(long version, @Nullable TableStatistics stats, boolean hasConsuming) {
+ _version = version;
+ _stats = stats;
+ _hasConsuming = hasConsuming;
+ }
+ }
+
+ /// A stored segment row plus the wall-clock time it was written, which the SQLite schema keeps in
+ /// its `updated_at_ms` column and exposes as [TableStatistics#getUpdatedAtMs()].
+ private static final class StoredSegment {
+ private final SegmentStatsRow _row;
+ private final long _updatedAtMs;
+
+ StoredSegment(SegmentStatsRow row, long updatedAtMs) {
+ _row = row;
+ _updatedAtMs = updatedAtMs;
+ }
+ }
+}
diff --git a/pinot-broker/src/main/java/org/apache/pinot/broker/stats/SqliteStatsStore.java b/pinot-broker/src/main/java/org/apache/pinot/broker/stats/SqliteStatsStore.java
new file mode 100644
index 00000000000..2b61c6370ad
--- /dev/null
+++ b/pinot-broker/src/main/java/org/apache/pinot/broker/stats/SqliteStatsStore.java
@@ -0,0 +1,885 @@
+/**
+ * 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.pinot.broker.stats;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import javax.annotation.Nullable;
+import org.apache.pinot.query.planner.spi.stats.ColumnStatistics;
+import org.apache.pinot.query.planner.spi.stats.ColumnValueType;
+import org.apache.pinot.query.planner.spi.stats.SegmentColumnStatsRow;
+import org.apache.pinot.query.planner.spi.stats.SegmentStatsRow;
+import org.apache.pinot.query.planner.spi.stats.StatConfidence;
+import org.apache.pinot.query.planner.spi.stats.StatsAggregations;
+import org.apache.pinot.query.planner.spi.stats.StatsStore;
+import org.apache.pinot.query.planner.spi.stats.StatsStoreException;
+import org.apache.pinot.query.planner.spi.stats.TableStatistics;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// SQLite-backed implementation of [StatsStore].
+///
+/// ### Threading model
+/// Uses a single shared writer [Connection] guarded by `synchronized` on
+/// `_writeLock`, plus a small pool (the configured pool size) of read-only connections
+/// served via a blocking queue. Reads from multiple threads proceed concurrently (each
+/// borrows a connection from the pool, uses it, then returns it). The writer connection sets
+/// `PRAGMA journal_mode=WAL` so readers never block writers.
+///
+/// ### Corruption handling
+/// [#init()] attempts to open the database and apply its schema. On any failure it logs a
+/// warning, deletes the DB file and its WAL/SHM siblings, then retries once from scratch.
+/// Only a second consecutive failure is propagated as [StatsStoreException].
+public class SqliteStatsStore implements StatsStore {
+ private static final Logger LOGGER = LoggerFactory.getLogger(SqliteStatsStore.class);
+
+ /// Default number of pooled readers when the operator sets none.
+ static final int DEFAULT_READ_POOL_SIZE = 4;
+
+ /// Default wait for a pooled reader. Deliberately short: this sits on the query-planning path,
+ /// where a late estimate is worth less than a fast plan without one, and the caller degrades to
+ /// heuristics rather than failing.
+ static final long DEFAULT_READ_BORROW_TIMEOUT_MS = 50;
+
+ private static final String DB_FILE_NAME = "broker-stats.sqlite";
+
+ /// Bump when [#SCHEMA_DDL] changes. Note that the persisted vocabulary is wider than the DDL:
+ /// `value_type` stores [ColumnValueType] names, which a newer build may extend without touching
+ /// the DDL. That case does not need a bump, because an unrecognized name resolves to `null` and
+ /// degrades to untrusted bounds rather than failing the read.
+ ///
+ /// There is no migration path on purpose: every row here is
+ /// derived from the ZooKeeper segment metadata the broker re-reads at startup, so a store whose
+ /// schema does not match is discarded and rebuilt rather than migrated. That keeps schema
+ /// changes, corruption and an unreadable file on one recovery path.
+ private static final int SCHEMA_VERSION = 2;
+
+ private static final String[] SCHEMA_DDL = {
+ "CREATE TABLE segment_stats ("
+ + "table_name TEXT NOT NULL, segment_name TEXT NOT NULL, crc INTEGER NOT NULL, "
+ + "total_docs INTEGER NOT NULL, size_bytes INTEGER NOT NULL, start_time_ms INTEGER NOT NULL, "
+ + "end_time_ms INTEGER NOT NULL, consuming INTEGER NOT NULL, updated_at_ms INTEGER NOT NULL, "
+ + "PRIMARY KEY (table_name, segment_name))",
+ "CREATE TABLE segment_col_stats ("
+ + "table_name TEXT NOT NULL, segment_name TEXT NOT NULL, column_name TEXT NOT NULL, "
+ + "ndv INTEGER NOT NULL, min_value TEXT, max_value TEXT, min_trusted INTEGER NOT NULL, "
+ + "avg_bytes REAL NOT NULL, null_fraction REAL NOT NULL, value_type TEXT, "
+ + "updated_at_ms INTEGER NOT NULL, "
+ + "PRIMARY KEY (table_name, segment_name, column_name))",
+ // getColumnStats filters by (table_name, column_name); column_name is the third key component
+ // so the primary key cannot serve it.
+ "CREATE INDEX idx_segment_col_stats_col ON segment_col_stats(table_name, column_name)",
+ "CREATE INDEX idx_segment_stats_time ON segment_stats(table_name, start_time_ms, end_time_ms)",
+ };
+
+ // SQL constants
+ private static final String SQL_UPSERT_SEGMENT =
+ "INSERT INTO segment_stats(table_name,segment_name,crc,total_docs,size_bytes,"
+ + "start_time_ms,end_time_ms,consuming,updated_at_ms) VALUES(?,?,?,?,?,?,?,?,?) "
+ + "ON CONFLICT(table_name,segment_name) DO UPDATE SET "
+ + "crc=excluded.crc,total_docs=excluded.total_docs,size_bytes=excluded.size_bytes,"
+ + "start_time_ms=excluded.start_time_ms,end_time_ms=excluded.end_time_ms,"
+ + "consuming=excluded.consuming,updated_at_ms=excluded.updated_at_ms";
+
+ private static final String SQL_UPSERT_COL =
+ "INSERT INTO segment_col_stats(table_name,segment_name,column_name,ndv,min_value,"
+ + "max_value,min_trusted,avg_bytes,null_fraction,value_type,updated_at_ms) "
+ + "VALUES(?,?,?,?,?,?,?,?,?,?,?) "
+ + "ON CONFLICT(table_name,segment_name,column_name) DO UPDATE SET "
+ + "ndv=excluded.ndv,min_value=excluded.min_value,max_value=excluded.max_value,"
+ + "min_trusted=excluded.min_trusted,avg_bytes=excluded.avg_bytes,"
+ + "null_fraction=excluded.null_fraction,value_type=excluded.value_type,"
+ + "updated_at_ms=excluded.updated_at_ms";
+
+ private static final String SQL_DELETE_SEGMENT =
+ "DELETE FROM segment_stats WHERE table_name=? AND segment_name=?";
+
+ private static final String SQL_DELETE_COL =
+ "DELETE FROM segment_col_stats WHERE table_name=? AND segment_name=?";
+
+ private static final String SQL_GET_CRCS =
+ "SELECT segment_name,crc FROM segment_stats WHERE table_name=?";
+
+ /// Rollup and the consuming flag in one scan: both are asked for on every plan, and splitting
+ /// them into two statements walked the same rows twice.
+ private static final String SQL_TABLE_STATS =
+ "SELECT SUM(CASE WHEN consuming=0 THEN total_docs ELSE 0 END),"
+ + "SUM(CASE WHEN consuming=0 THEN size_bytes ELSE 0 END),"
+ + "MAX(CASE WHEN consuming=0 THEN updated_at_ms ELSE 0 END),"
+ + "SUM(CASE WHEN consuming=0 THEN 1 ELSE 0 END),"
+ + "MAX(consuming) FROM segment_stats WHERE table_name=?";
+
+ private static final String SQL_COL_STATS =
+ "SELECT s.total_docs,c.ndv,c.min_value,c.max_value,c.min_trusted,c.avg_bytes,c.null_fraction,"
+ + "c.value_type "
+ + "FROM segment_col_stats c "
+ + "JOIN segment_stats s ON s.table_name=c.table_name AND s.segment_name=c.segment_name "
+ + "WHERE c.table_name=? AND c.column_name=? AND s.consuming=0";
+
+ // The overlap predicate mirrors the Java-side check in estimateRowsInTimeRange: it prunes
+ // segments that cannot overlap [startMs, endMs) inside SQLite instead of materializing them
+ // over JDBC only to be skipped. Segments with unknown times (-1 sentinels) must be retained —
+ // they are included conservatively by the Java logic.
+ //
+ // The UNION ALL arm emits a sentinel row (start_time_ms = -2, an otherwise impossible value)
+ // whenever the table has any committed stats at all. It lets the caller distinguish "no segment
+ // overlaps the range" (a real estimate of 0) from "no stats for this table" (empty) within a
+ // SINGLE statement — i.e. a single consistent snapshot; splitting this into a second existence
+ // query would race with concurrent stats writes.
+ private static final String SQL_TIME_RANGE =
+ "SELECT total_docs,start_time_ms,end_time_ms "
+ + "FROM segment_stats WHERE table_name=? AND consuming=0 "
+ + "AND (start_time_ms=-1 OR end_time_ms=-1 OR (end_time_ms>? AND start_time_ms<?)) "
+ + "UNION ALL SELECT -1,-2,-2 WHERE EXISTS("
+ + "SELECT 1 FROM segment_stats WHERE table_name=? AND consuming=0)";
+
+ private static final String SQL_HAS_CONSUMING =
+ "SELECT 1 FROM segment_stats WHERE table_name=? AND consuming=1 LIMIT 1";
+
+ // Both tables are consulted: a table could in principle have column rows without segment rows.
+ private static final String SQL_LIST_TABLES =
+ "SELECT table_name FROM segment_stats UNION SELECT table_name FROM segment_col_stats";
+
+ private static final String SQL_PURGE_TABLE_SEG =
+ "DELETE FROM segment_stats WHERE table_name=?";
+
+ private static final String SQL_PURGE_TABLE_COL =
+ "DELETE FROM segment_col_stats WHERE table_name=?";
+
+ private static final String SQL_PURGE_ALL_SEG = "DELETE FROM segment_stats";
+ private static final String SQL_PURGE_ALL_COL = "DELETE FROM segment_col_stats";
+
+ private final Path _dbDirectory;
+ private final Path _dbPath;
+
+ private final int _readPoolSize;
+ private final long _readBorrowTimeoutMs;
+
+ /// Write connection — all mutations go through this; guarded by _writeLock.
+ /// Volatile because [#openConnections()] and [#closeConnections()] assign it outside the
+ /// `_writeLock` that guards every other access, during init and rebuild.
+ private volatile Connection _writeConn;
+ /// Serialises writers. Needed even though SQLite is itself thread-safe, because a write here is
+ /// a multi-statement transaction (`addBatch` / `executeBatch` / `commit`) and the transaction
+ /// boundary is per-CONNECTION state: two threads interleaving on this shared connection would
+ /// let one thread's `commit()` publish another's half-written batch.
+ ///
+ /// It does NOT make readers wait. The database runs in WAL mode, so a writer never blocks a
+ /// reader and a reader never blocks the writer; this lock is only ever contended between
+ /// concurrent writers.
+ private final Object _writeLock = new Object();
+
+ /// Pool of read-only connections. Each reader borrows a connection, uses it,
+ /// then returns it via `offer()`. Sized at the configured pool size.
+ private final ArrayBlockingQueue<PooledReader> _readPool;
+
+ /// Per-table write counter backing the rollup cache; see [#aggregate(String)].
+ private final Map<String, AtomicLong> _versions = new ConcurrentHashMap<>();
+
+ /// Cached per-table rollups, each stamped with the [#_versions] value it was computed from.
+ private final Map<String, CachedAggregate> _aggregates = new ConcurrentHashMap<>();
+
+ private volatile boolean _closed = false;
+
+ /// Constructs a new `SqliteStatsStore` that stores its database in the given directory.
+ /// The database file will be `<dbDirectory>/broker-stats.sqlite`.
+ ///
+ /// @param dbDirectory directory in which to store the database file; created if absent
+ public SqliteStatsStore(Path dbDirectory) {
+ this(dbDirectory, DEFAULT_READ_POOL_SIZE, DEFAULT_READ_BORROW_TIMEOUT_MS);
+ }
+
+ /// @param dbDirectory directory in which to store the database file; created if absent
+ /// @param readPoolSize number of pooled readers; the pool is a bound, so this caps how
+ /// many estimates can be served concurrently
+ /// @param readBorrowTimeoutMs how long a reader waits for a pooled connection before giving up
+ /// and reporting no statistics
+ public SqliteStatsStore(Path dbDirectory, int readPoolSize, long readBorrowTimeoutMs) {
+ _dbDirectory = dbDirectory;
+ _dbPath = dbDirectory.resolve(DB_FILE_NAME);
+ _readPoolSize = readPoolSize > 0 ? readPoolSize : DEFAULT_READ_POOL_SIZE;
+ _readBorrowTimeoutMs = readBorrowTimeoutMs > 0 ? readBorrowTimeoutMs : DEFAULT_READ_BORROW_TIMEOUT_MS;
+ _readPool = new ArrayBlockingQueue<>(_readPoolSize);
+ }
+
+ // ---------------------------------------------------------------------------
+ // Lifecycle
+ // ---------------------------------------------------------------------------
+
+ @Override
+ public void init()
+ throws StatsStoreException {
+ try {
+ openAndApplySchema();
+ } catch (Exception firstEx) {
+ LOGGER.warn(
+ "Failed to open stats store at {}; deleting and retrying from scratch. Cause: {}",
+ _dbPath, firstEx.getMessage(), firstEx);
+ closeConnections();
+ deleteDbFiles();
+ try {
+ openAndApplySchema();
+ } catch (Exception secondEx) {
+ throw new StatsStoreException(
+ "Cannot initialise SqliteStatsStore at " + _dbPath, secondEx);
+ }
+ }
+ }
+
+ private void openAndApplySchema()
+ throws Exception {
+ try {
+ openConnections();
+ } catch (Exception e) {
+ // Leave nothing bound to a database init() may be about to delete: a surviving connection
+ // would keep the old inode alive and serve reads from a file nothing writes to any more.
+ closeConnections();
+ throw e;
+ }
+ }
+
+ private void openConnections()
+ throws Exception {
+ Files.createDirectories(_dbDirectory);
+ String jdbcUrl = "jdbc:sqlite:" + _dbPath.toAbsolutePath();
+
+ // Open the shared writer connection.
+ // Set WAL and synchronous PRAGMAs with autoCommit=true (WAL mode change cannot be done
+ // inside a transaction), then switch to manual-commit mode for subsequent writes.
+ Connection conn = DriverManager.getConnection(jdbcUrl);
+ try (Statement st = conn.createStatement()) {
+ st.execute("PRAGMA journal_mode=WAL");
+ st.execute("PRAGMA synchronous=NORMAL");
+ }
+ applySchema(conn);
+ conn.setAutoCommit(false);
+ _writeConn = conn;
+
+ // Open read-only connections for the pool
+ for (int i = 0; i < _readPoolSize; i++) {
+ Connection rConn = DriverManager.getConnection(jdbcUrl);
+ rConn.setAutoCommit(true);
+ try (Statement st = rConn.createStatement()) {
+ st.execute("PRAGMA journal_mode=WAL");
+ st.execute("PRAGMA synchronous=NORMAL");
+ }
+ PooledReader reader = new PooledReader(rConn);
+ if (!_readPool.offer(reader)) {
+ reader.close();
+ }
+ }
+ }
+
+ /// Closes and forgets every connection this store holds. Safe to call repeatedly.
+ private void closeConnections() {
+ // A rebuild drops the file the cached rollups were computed from, so they cannot outlive it.
+ _aggregates.clear();
+ _versions.clear();
+ closeQuietly(_writeConn);
+ _writeConn = null;
+ PooledReader pooled;
+ while ((pooled = _readPool.poll()) != null) {
+ // Closes the cached statements as well as the connection.
+ pooled.close();
+ }
+ }
+
+ /// Creates the schema on an empty database, or verifies that an existing one matches
+ /// [#SCHEMA_VERSION].
+ ///
+ /// SQLite's own `user_version` pragma carries the version, so no bookkeeping table and no
+ /// migration library is needed for what is a derived, rebuildable cache. A mismatch throws, which
+ /// [#init()] turns into delete-and-recreate.
+ private static void applySchema(Connection conn)
+ throws SQLException {
+ try (Statement st = conn.createStatement()) {
+ int version;
+ try (ResultSet rs = st.executeQuery("PRAGMA user_version")) {
+ version = rs.next() ? rs.getInt(1) : 0;
+ }
+ if (version == SCHEMA_VERSION) {
+ return;
+ }
+ if (version != 0) {
+ throw new SQLException("Stats store schema version " + version + " does not match expected "
+ + SCHEMA_VERSION + "; the store will be rebuilt");
+ }
+ for (String ddl : SCHEMA_DDL) {
+ st.execute(ddl);
+ }
+ st.execute("PRAGMA user_version=" + SCHEMA_VERSION);
+ }
+ }
+
+ /// Deletes the SQLite DB file and its WAL / SHM siblings if they exist.
+ private void deleteDbFiles() {
+ tryDelete(_dbPath);
+ tryDelete(_dbDirectory.resolve(DB_FILE_NAME + "-wal"));
+ tryDelete(_dbDirectory.resolve(DB_FILE_NAME + "-shm"));
+ }
+
+ private static void tryDelete(Path p) {
+ try {
+ Files.deleteIfExists(p);
+ } catch (IOException e) {
+ LOGGER.warn("Could not delete {}: {}", p, e.getMessage());
+ }
+ }
+
+ @Override
+ public void close() {
+ _closed = true;
+ synchronized (_writeLock) {
+ closeQuietly(_writeConn);
+ _writeConn = null;
+ }
+ PooledReader r;
+ while ((r = _readPool.poll()) != null) {
+ r.close();
+ }
+ }
+
+ /// Returns the writer, or throws if the store closed between [#checkOpen()] and acquiring the
+ /// lock. Without this a writer that passed the flag check would NPE on a nulled connection, and
+ /// an NPE is neither rolled back here nor caught by the listeners that call this.
+ private Connection writeConnection()
+ throws StatsStoreException {
+ if (_writeConn == null) {
+ throw new StatsStoreException("SqliteStatsStore is closed");
+ }
+ return _writeConn;
+ }
+
+ private static void closeQuietly(@Nullable Connection conn) {
+ if (conn != null) {
+ try {
+ conn.close();
+ } catch (SQLException e) {
+ LOGGER.debug("Error closing connection", e);
+ }
+ }
+ }
+
+ // ---------------------------------------------------------------------------
+ // Write operations
+ // ---------------------------------------------------------------------------
+
+ @Override
+ public void upsertSegmentStats(String tableNameWithType, List<SegmentStatsRow> rows)
+ throws StatsStoreException {
+ checkOpen();
+ long now = System.currentTimeMillis();
+ synchronized (_writeLock) {
+ try {
+ try (PreparedStatement ps = writeConnection().prepareStatement(SQL_UPSERT_SEGMENT)) {
+ for (SegmentStatsRow row : rows) {
+ ps.setString(1, tableNameWithType);
+ ps.setString(2, row.segmentName());
+ ps.setLong(3, row.crc());
+ ps.setLong(4, row.totalDocs());
+ ps.setLong(5, row.sizeBytes());
+ ps.setLong(6, row.startTimeMs());
+ ps.setLong(7, row.endTimeMs());
+ ps.setInt(8, row.consuming() ? 1 : 0);
+ ps.setLong(9, now);
+ ps.addBatch();
+ }
+ ps.executeBatch();
+ }
+ writeConnection().commit();
+ invalidate(tableNameWithType);
+ } catch (SQLException e) {
+ rollbackQuietly(_writeConn);
+ invalidate(tableNameWithType);
+ throw new StatsStoreException("upsertSegmentStats failed for " + tableNameWithType, e);
+ }
+ }
+ }
+
+ @Override
+ public void upsertSegmentColumnStats(String tableNameWithType, List<SegmentColumnStatsRow> rows)
+ throws StatsStoreException {
+ checkOpen();
+ long now = System.currentTimeMillis();
+ synchronized (_writeLock) {
+ try {
+ try (PreparedStatement ps = writeConnection().prepareStatement(SQL_UPSERT_COL)) {
+ for (SegmentColumnStatsRow row : rows) {
+ ps.setString(1, tableNameWithType);
+ ps.setString(2, row.segmentName());
+ ps.setString(3, row.columnName());
+ ps.setLong(4, row.ndv());
+ ps.setString(5, row.minValue());
+ ps.setString(6, row.maxValue());
+ ps.setInt(7, row.minTrusted() ? 1 : 0);
+ ps.setDouble(8, row.avgBytesPerValue());
+ ps.setDouble(9, row.nullFraction());
+ ps.setString(10, row.valueType() == null ? null : row.valueType().name());
+ ps.setLong(11, now);
+ ps.addBatch();
+ }
+ ps.executeBatch();
+ }
+ writeConnection().commit();
+ } catch (SQLException e) {
+ rollbackQuietly(_writeConn);
+ throw new StatsStoreException(
+ "upsertSegmentColumnStats failed for " + tableNameWithType, e);
+ }
+ }
+ }
+
+ @Override
+ public void removeSegments(String tableNameWithType, Collection<String> segmentNames)
+ throws StatsStoreException {
+ checkOpen();
+ if (segmentNames.isEmpty()) {
+ return;
+ }
+ synchronized (_writeLock) {
+ try {
+ try (PreparedStatement psSeg = writeConnection().prepareStatement(SQL_DELETE_SEGMENT);
+ PreparedStatement psCol = writeConnection().prepareStatement(SQL_DELETE_COL)) {
+ for (String seg : segmentNames) {
+ psSeg.setString(1, tableNameWithType);
+ psSeg.setString(2, seg);
+ psSeg.addBatch();
+ psCol.setString(1, tableNameWithType);
+ psCol.setString(2, seg);
+ psCol.addBatch();
+ }
+ psSeg.executeBatch();
+ psCol.executeBatch();
+ }
+ writeConnection().commit();
+ invalidate(tableNameWithType);
+ } catch (SQLException e) {
+ rollbackQuietly(_writeConn);
+ invalidate(tableNameWithType);
+ throw new StatsStoreException("removeSegments failed for " + tableNameWithType, e);
+ }
+ }
+ }
+
+ @Override
+ public boolean hasConsumingSegments(String tableNameWithType)
+ throws StatsStoreException {
+ return aggregate(tableNameWithType)._hasConsuming;
+ }
+
+
+ @Override
+ public Set<String> getTables()
+ throws StatsStoreException {
+ checkOpen();
+ PooledReader reader = borrowReader();
+ try {
+ Set<String> tables = new HashSet<>();
+ PreparedStatement ps = reader.statement(SQL_LIST_TABLES);
+ try (ResultSet rs = ps.executeQuery()) {
+ while (rs.next()) {
+ tables.add(rs.getString(1));
+ }
+ }
+ return tables;
+ } catch (SQLException e) {
+ throw new StatsStoreException("getTables failed", e);
+ } finally {
+ returnReader(reader);
+ }
+ }
+
+ @Override
+ public void purgeTable(String tableNameWithType)
+ throws StatsStoreException {
+ checkOpen();
+ synchronized (_writeLock) {
+ try {
+ try (PreparedStatement psSeg = writeConnection().prepareStatement(SQL_PURGE_TABLE_SEG);
+ PreparedStatement psCol = writeConnection().prepareStatement(SQL_PURGE_TABLE_COL)) {
+ psSeg.setString(1, tableNameWithType);
+ psSeg.executeUpdate();
+ psCol.setString(1, tableNameWithType);
+ psCol.executeUpdate();
+ }
+ writeConnection().commit();
+ // Drop the cache entry outright rather than only stamping it: nothing is left to roll up.
+ _aggregates.remove(tableNameWithType);
+ _versions.remove(tableNameWithType);
+ } catch (SQLException e) {
+ rollbackQuietly(_writeConn);
+ invalidate(tableNameWithType);
+ throw new StatsStoreException("purgeTable failed for " + tableNameWithType, e);
+ }
+ }
+ }
+
+ @Override
+ public void purgeAll()
+ throws StatsStoreException {
+ checkOpen();
+ synchronized (_writeLock) {
+ try {
+ try (Statement st = writeConnection().createStatement()) {
+ st.execute(SQL_PURGE_ALL_SEG);
+ st.execute(SQL_PURGE_ALL_COL);
+ }
+ writeConnection().commit();
+ _aggregates.clear();
+ _versions.clear();
+ } catch (SQLException e) {
+ rollbackQuietly(_writeConn);
+ invalidateAll();
+ throw new StatsStoreException("purgeAll failed", e);
+ }
+ }
+ }
+
+ // ---------------------------------------------------------------------------
+ // Read operations
+ // ---------------------------------------------------------------------------
+
+ @Override
+ public Map<String, Long> getSegmentCrcs(String tableNameWithType)
+ throws StatsStoreException {
+ checkOpen();
+ PooledReader reader = borrowReader();
+ try {
+ Map<String, Long> result = new HashMap<>();
+ PreparedStatement ps = reader.statement(SQL_GET_CRCS);
+ {
+ ps.setString(1, tableNameWithType);
+ try (ResultSet rs = ps.executeQuery()) {
+ while (rs.next()) {
+ result.put(rs.getString(1), rs.getLong(2));
+ }
+ }
+ }
+ return result;
+ } catch (SQLException e) {
+ throw new StatsStoreException("getSegmentCrcs failed for " + tableNameWithType, e);
+ } finally {
+ returnReader(reader);
+ }
+ }
+
+ @Override
+ @Nullable
+ public TableStatistics getTableStats(String tableNameWithType)
+ throws StatsStoreException {
+ return aggregate(tableNameWithType)._stats;
+ }
+
+
+ /// Returns per-column statistics aggregated across all non-consuming segments for the given
+ /// table and column, or `null` if no rows exist.
+ ///
+ /// #### NDV
+ /// Returned as `MAX(ndv)` over segments with [StatConfidence#ESTIMATED]. The
+ /// true value lies in `[MAX(ndv), min(SUM(ndv), tableRowCount)]`; we report the lower
+ /// bound (MAX) because the upper bound is not representable as a single value in the contract.
+ ///
+ /// #### Min/Max
+ /// Compared numerically when both values parse as [Double], else lexically. Comparison
+ /// is done in Java (not SQL) to avoid SQLite TEXT-affinity ordering issues (e.g. "9" > "10").
+ @Override
+ @Nullable
+ public ColumnStatistics getColumnStats(String tableNameWithType, String columnName)
+ throws StatsStoreException {
+ checkOpen();
+ PooledReader reader = borrowReader();
+ try {
+ StatsAggregations.ColumnStatsAccumulator accumulator = new StatsAggregations.ColumnStatsAccumulator();
+ PreparedStatement ps = reader.statement(SQL_COL_STATS);
+ {
+ ps.setString(1, tableNameWithType);
+ ps.setString(2, columnName);
+ try (ResultSet rs = ps.executeQuery()) {
+ while (rs.next()) {
+ accumulator.add(rs.getLong(1), SegmentColumnStatsRow.builder()
+ .segmentName("")
+ .columnName(columnName)
+ .ndv(rs.getLong(2))
+ .bounds(rs.getString(3), rs.getString(4), ColumnValueType.fromName(rs.getString(8)))
+ .minTrusted(rs.getInt(5) != 0)
+ .avgBytesPerValue(rs.getDouble(6))
+ .nullFraction(rs.getDouble(7))
+ .build());
+ }
+ }
+ }
+ return accumulator.isEmpty() ? null : accumulator.build(columnName);
+ } catch (SQLException e) {
+ throw new StatsStoreException(
+ "getColumnStats failed for " + tableNameWithType + "." + columnName, e);
+ } finally {
+ returnReader(reader);
+ }
+ }
+
+ @Override
+ public OptionalLong estimateRowsInTimeRange(String tableNameWithType, long startMs, long endMs)
+ throws StatsStoreException {
+ checkOpen();
+ PooledReader reader = borrowReader();
+ try {
+ long totalRows = 0;
+ boolean hasAnyRow = false;
+
+ PreparedStatement ps = reader.statement(SQL_TIME_RANGE);
+ {
+ ps.setString(1, tableNameWithType);
+ ps.setLong(2, startMs);
+ ps.setLong(3, endMs);
+ ps.setString(4, tableNameWithType);
+ try (ResultSet rs = ps.executeQuery()) {
+ while (rs.next()) {
+ long docs = rs.getLong(1);
+ long segStart = rs.getLong(2);
+ long segEnd = rs.getLong(3);
+
+ hasAnyRow = true;
+
+ // Existence sentinel (see SQL_TIME_RANGE): committed stats exist, contributes 0 rows
+ if (segStart == -2) {
+ continue;
+ }
+
+ // The SQL predicate already prunes non-overlapping segments; the shared helper decides
+ // how much of a surviving segment counts (see StatsAggregations#overlapRows).
+ totalRows += StatsAggregations.overlapRows(docs, segStart, segEnd, startMs, endMs);
+ }
+ }
+ }
+
+ return hasAnyRow ? OptionalLong.of(totalRows) : OptionalLong.empty();
+ } catch (SQLException e) {
+ throw new StatsStoreException(
+ "estimateRowsInTimeRange failed for " + tableNameWithType, e);
+ } finally {
+ returnReader(reader);
+ }
+ }
+
+ // ---------------------------------------------------------------------------
+ // Helpers
+ // ---------------------------------------------------------------------------
+
+ private void checkOpen()
+ throws StatsStoreException {
+ if (_closed) {
+ throw new StatsStoreException("SqliteStatsStore is closed");
+ }
+ }
+
+ private PooledReader borrowReader()
+ throws StatsStoreException {
+ PooledReader reader = _readPool.poll();
+ if (reader != null) {
+ return reader;
+ }
+ // The pool is a bound, not a hint. Opening a connection per waiting reader would make the
+ // exhausted case -- the normal one on a broker with more planning threads than pooled
+ // connections -- the most expensive path: every read would pay a full database open plus its
+ // own page cache, and nothing would cap the file descriptors a burst can hold open. Waiting
+ // briefly instead keeps the cost bounded, and the caller already degrades to no statistics
+ // rather than failing a query.
+ try {
+ reader = _readPool.poll(_readBorrowTimeoutMs, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new StatsStoreException("Interrupted while waiting for a read connection", e);
+ }
+ if (reader == null) {
+ throw new StatsStoreException(
+ "Timed out after " + _readBorrowTimeoutMs + "ms waiting for one of " + _readPoolSize
+ + " read connections");
+ }
+ // A closed store must not hand out a connection it is in the middle of tearing down.
+ checkOpen();
+ return reader;
+ }
+
+ private void returnReader(@Nullable PooledReader reader) {
+ if (reader == null) {
+ return;
+ }
+ // Do not re-pool into a store that closed while this read was in flight, or close() would
+ // leave a live connection -- and its statements -- behind.
+ if (_closed || !_readPool.offer(reader)) {
+ reader.close();
+ }
+ }
+
+ /// A pooled read connection together with the statements compiled on it.
+ ///
+ /// SQLite compiles each statement into a VDBE program, and that compilation runs the query
+ /// planner -- so re-preparing on every call re-does index selection. Measured on this schema:
+ /// preparing is ~58% of total query cost for a table with 50 segments, where compilation
+ /// dominates execution, falling to ~4% at 1000 segments. Most tables sit at the small end and
+ /// the planner issues several of these per compile, so the statements are kept with their
+ /// connection rather than rebuilt.
+ ///
+ /// Not thread-safe, and does not need to be: an instance is owned by exactly one thread between
+ /// [#borrowReader()] and [#returnReader].
+ private static final class PooledReader {
+ private final Connection _conn;
+ private final Map<String, PreparedStatement> _statements = new HashMap<>();
+
+ PooledReader(Connection conn) {
+ _conn = conn;
+ }
+
+ /// The statement for `sql`, compiled on first use and reused afterwards, with any parameters
+ /// from the previous execution cleared.
+ PreparedStatement statement(String sql)
+ throws SQLException {
+ PreparedStatement ps = _statements.get(sql);
+ if (ps == null) {
+ ps = _conn.prepareStatement(sql);
+ _statements.put(sql, ps);
+ }
+ ps.clearParameters();
+ return ps;
+ }
+
+ void close() {
+ for (PreparedStatement ps : _statements.values()) {
+ try {
+ ps.close();
+ } catch (SQLException e) {
+ LOGGER.debug("Closing pooled statement failed", e);
+ }
+ }
+ _statements.clear();
+ closeQuietly(_conn);
+ }
+ }
+
+ private static void rollbackQuietly(@Nullable Connection conn) {
+ if (conn == null) {
+ return;
+ }
+ try {
+ conn.rollback();
+ } catch (SQLException e) {
+ LOGGER.debug("Rollback failed", e);
+ }
+ }
+
+ /// Returns the cached rollup for a table, recomputing it only when a write has landed since the
+ /// cached copy was taken.
+ ///
+ /// Query planning asks for these on every compile while writes arrive at segment-push cadence,
+ /// so without a cache every plan would scan all of a table's segment rows -- on a table with
+ /// hundreds of thousands of segments that lands directly in planning latency. The version stamp
+ /// is what makes caching safe without holding the write lock: a rollup computed from rows that
+ /// changed underneath it is used for this call but not published.
+ private CachedAggregate aggregate(String tableNameWithType)
+ throws StatsStoreException {
+ AtomicLong version = _versions.computeIfAbsent(tableNameWithType, k -> new AtomicLong());
+ long observed = version.get();
+ CachedAggregate cached = _aggregates.get(tableNameWithType);
+ if (cached != null && cached._version == observed) {
+ return cached;
+ }
+ CachedAggregate computed = computeAggregate(tableNameWithType, observed);
+ if (version.get() == observed) {
+ _aggregates.put(tableNameWithType, computed);
+ }
+ return computed;
+ }
+
+ private CachedAggregate computeAggregate(String tableNameWithType, long version)
+ throws StatsStoreException {
+ checkOpen();
+ PooledReader reader = borrowReader();
+ try {
+ PreparedStatement ps = reader.statement(SQL_TABLE_STATS);
+ ps.setString(1, tableNameWithType);
+ try (ResultSet rs = ps.executeQuery()) {
+ // Aggregates always return one row, holding SQL NULLs when the table has none.
+ if (!rs.next()) {
+ return new CachedAggregate(version, null, false);
+ }
+ long totalDocs = rs.getLong(1);
+ long sizeBytes = rs.getLong(2);
+ long maxUpdatedAt = rs.getLong(3);
+ long committed = rs.getLong(4);
+ boolean hasConsuming = rs.getLong(5) == 1;
+ if (committed == 0) {
+ return new CachedAggregate(version, null, hasConsuming);
+ }
+ return new CachedAggregate(version, TableStatistics.builder()
+ .rowCount(totalDocs, StatConfidence.EXACT)
+ .tableSizeBytes(sizeBytes, StatConfidence.EXACT)
+ .updatedAtMs(maxUpdatedAt)
+ .build(), hasConsuming);
+ }
+ } catch (SQLException e) {
+ throw new StatsStoreException("getTableStats failed for " + tableNameWithType, e);
+ } finally {
+ returnReader(reader);
+ }
+ }
+
+ /// Marks a table's cached rollup stale. Called from the write paths, which hold `_writeLock`.
+ private void invalidate(String tableNameWithType) {
+ _versions.computeIfAbsent(tableNameWithType, k -> new AtomicLong()).incrementAndGet();
+ _aggregates.remove(tableNameWithType);
+ }
+
+ /// Marks every table's cached rollup stale, for writes that span tables.
+ private void invalidateAll() {
+ _aggregates.clear();
+ _versions.values().forEach(AtomicLong::incrementAndGet);
+ }
+
+ private static final class CachedAggregate {
+ private final long _version;
+ @Nullable
+ private final TableStatistics _stats;
+ private final boolean _hasConsuming;
+
+ CachedAggregate(long version, @Nullable TableStatistics stats, boolean hasConsuming) {
+ _version = version;
+ _stats = stats;
+ _hasConsuming = hasConsuming;
+ }
+ }
+}
diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/stats/InMemoryStatsStoreTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/stats/InMemoryStatsStoreTest.java
new file mode 100644
index 00000000000..c55086c390b
--- /dev/null
+++ b/pinot-broker/src/test/java/org/apache/pinot/broker/stats/InMemoryStatsStoreTest.java
@@ -0,0 +1,52 @@
+/**
+ * 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.pinot.broker.stats;
+
+import java.util.List;
+import org.apache.pinot.query.planner.spi.stats.SegmentStatsRow;
+import org.apache.pinot.query.planner.spi.stats.StatsStore;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertTrue;
+
+
+/// Runs the shared [StatsStore] contract against [InMemoryStatsStore], plus the one behavior that
+/// distinguishes it: a new instance starts empty, so the caller re-collects everything.
+public class InMemoryStatsStoreTest extends StatsStoreContractTest {
+
+ @Override
+ protected StatsStore createStore() {
+ return new InMemoryStatsStore();
+ }
+
+ /// A fresh store keeps nothing from a previous one. Reconciliation relies on this: an empty
+ /// crc map makes every segment look new, so the listener re-upserts the lot.
+ @Test
+ public void testNothingSurvivesANewInstance()
+ throws Exception {
+ _store.upsertSegmentStats("myTable_OFFLINE",
+ List.of(new SegmentStatsRow("seg1", 42L, 500L, 2000L, 0L, 100L, false)));
+ assertTrue(_store.getSegmentCrcs("myTable_OFFLINE").containsKey("seg1"));
+
+ try (StatsStore fresh = createStore()) {
+ fresh.init();
+ assertTrue(fresh.getSegmentCrcs("myTable_OFFLINE").isEmpty(), "A new store must start empty");
+ }
+ }
+}
diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/stats/SqliteStatsStoreTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/stats/SqliteStatsStoreTest.java
new file mode 100644
index 00000000000..e5362b488cf
--- /dev/null
+++ b/pinot-broker/src/test/java/org/apache/pinot/broker/stats/SqliteStatsStoreTest.java
@@ -0,0 +1,259 @@
+/**
+ * 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.pinot.broker.stats;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.Statement;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import org.apache.pinot.query.planner.spi.stats.SegmentStatsRow;
+import org.apache.pinot.query.planner.spi.stats.StatsStore;
+import org.apache.pinot.query.planner.spi.stats.StatsStoreException;
+import org.apache.pinot.query.planner.spi.stats.TableStatistics;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.*;
+
+
+/// Runs the shared [StatsStore] contract against [SqliteStatsStore], plus the cases that only a
+/// durable, file-backed store can have.
+public class SqliteStatsStoreTest extends StatsStoreContractTest {
+
+ private Path _tempDir;
+
+ @Override
+ protected StatsStore createStore()
+ throws Exception {
+ _tempDir = Files.createTempDirectory("sqlite-stats-test-");
+ return new SqliteStatsStore(_tempDir);
+ }
+
+ @Override
+ protected void cleanUp()
+ throws Exception {
+ deleteRecursively(_tempDir);
+ }
+
+ // ---------------------------------------------------------------------------
+ // Persistence across reopen
+ // ---------------------------------------------------------------------------
+
+ @Test
+ public void testPersistenceAcrossReopen()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("seg1", 42L, 500L, 2000L, 0L, 100L, false)
+ ));
+ _store.close();
+ _store = null;
+
+ // Open a new store on the same directory
+ SqliteStatsStore store2 = new SqliteStatsStore(_tempDir);
+ store2.init();
+ try {
+ Map<String, Long> crcs = store2.getSegmentCrcs(TABLE_A);
+ assertEquals(crcs.size(), 1);
+ assertEquals(crcs.get("seg1").longValue(), 42L);
+
+ TableStatistics stats = store2.getTableStats(TABLE_A);
+ assertNotNull(stats);
+ assertEquals(stats.getRowCount(), 500L);
+ } finally {
+ store2.close();
+ }
+ }
+
+ // ---------------------------------------------------------------------------
+ // Corruption recovery
+ // ---------------------------------------------------------------------------
+
+ @Test
+ public void testCorruptionRecovery()
+ throws Exception {
+ // Write some data, then close
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("seg1", 1L, 100L, 1000L, 0L, 10L, false)
+ ));
+ _store.close();
+ _store = null;
+
+ // Corrupt the DB file
+ Path dbFile = _tempDir.resolve("broker-stats.sqlite");
+ Files.write(dbFile, "this is not a valid sqlite database file!!!".getBytes());
+
+ // Opening a new store should recover silently
+ SqliteStatsStore store2 = new SqliteStatsStore(_tempDir);
+ store2.init(); // must not throw
+ try {
+ // After recovery, the store is empty
+ assertNull(store2.getTableStats(TABLE_A));
+ Map<String, Long> crcs = store2.getSegmentCrcs(TABLE_A);
+ assertTrue(crcs.isEmpty());
+ } finally {
+ store2.close();
+ }
+ }
+
+ /// The schema carries no migration path on purpose: rows here are derived from ZooKeeper and can
+ /// always be re-collected, so a store written by a different schema version is discarded rather
+ /// than migrated -- the same recovery path a corrupt file takes.
+ @Test
+ public void testSchemaVersionMismatchRebuilds()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(seg("seg1", 42L, 500L, 2000L, 0L, 100L, false)));
+ _store.close();
+ _store = null;
+
+ Path dbFile = _tempDir.resolve("broker-stats.sqlite");
+ try (Connection conn = DriverManager.getConnection("jdbc:sqlite:" + dbFile.toAbsolutePath());
+ Statement st = conn.createStatement()) {
+ st.execute("PRAGMA user_version=9999");
+ }
+
+ try (SqliteStatsStore reopened = new SqliteStatsStore(_tempDir)) {
+ reopened.init();
+ assertTrue(reopened.getSegmentCrcs(TABLE_A).isEmpty(), "A foreign schema version must be rebuilt empty");
+ assertNull(reopened.getTableStats(TABLE_A));
+ }
+ }
+
+ // ---------------------------------------------------------------------------
+ // Read pool sizing and borrow timeout
+ // ---------------------------------------------------------------------------
+
+ /// The pool is a concurrency bound, not a per-caller connection, so a store configured with a
+ /// single reader must still serve every caller correctly: readers are borrowed and returned,
+ /// never leaked and never handed to two callers at once.
+ @Test
+ public void testSingleReaderPoolStillServesEveryRead()
+ throws Exception {
+ Path dir = Files.createTempDirectory("sqlite-stats-pool-");
+ try (SqliteStatsStore store = new SqliteStatsStore(dir, 1, 5000)) {
+ store.init();
+ store.upsertSegmentStats(TABLE_A, List.of(seg("seg1", 1L, 500L, 2000L, 0L, 100L, false)));
+
+ // More reads than pooled readers: each must borrow, use and return the one connection.
+ for (int i = 0; i < 8; i++) {
+ TableStatistics stats = store.getTableStats(TABLE_A);
+ assertNotNull(stats, "Read " + i + " got no statistics");
+ assertEquals(stats.getRowCount(), 500L, "Read " + i + " saw the wrong row count");
+ }
+ } finally {
+ deleteRecursively(dir);
+ }
+ }
+
+ /// An exhausted pool must give up after the configured timeout rather than block a planner
+ /// thread indefinitely, and must recover once readers come back.
+ @Test
+ public void testBorrowTimesOutWhenPoolIsExhausted()
+ throws Exception {
+ Path dir = Files.createTempDirectory("sqlite-stats-timeout-");
+ long timeoutMs = 100;
+ try (SqliteStatsStore store = new SqliteStatsStore(dir, 2, timeoutMs)) {
+ store.init();
+ store.upsertSegmentStats(TABLE_A, List.of(seg("seg1", 1L, 500L, 2000L, 0L, 100L, false)));
+
+ // Hold every reader. Reflection rather than blocked threads: an empty pool is exactly the
+ // state under test, and reproducing it with concurrent reads would race against how fast
+ // SQLite answers.
+ Deque<Object> held = drainReadPool(store);
+ assertEquals(held.size(), 2, "Expected the configured number of pooled readers");
+ try {
+ long startNanos = System.nanoTime();
+ StatsStoreException e = expectThrows(StatsStoreException.class, () -> store.getTableStats(TABLE_A));
+ long elapsedMs = (System.nanoTime() - startNanos) / 1_000_000;
+
+ assertTrue(e.getMessage().contains("Timed out"), "Unexpected message: " + e.getMessage());
+ assertTrue(elapsedMs >= timeoutMs,
+ "Gave up after " + elapsedMs + "ms, before the configured " + timeoutMs + "ms");
+ } finally {
+ returnToReadPool(store, held);
+ }
+
+ // With the readers back, reads succeed again.
+ TableStatistics recovered = store.getTableStats(TABLE_A);
+ assertNotNull(recovered);
+ assertEquals(recovered.getRowCount(), 500L);
+ } finally {
+ deleteRecursively(dir);
+ }
+ }
+
+ private static Deque<Object> drainReadPool(SqliteStatsStore store)
+ throws Exception {
+ BlockingQueue<Object> pool = readPool(store);
+ Deque<Object> held = new ArrayDeque<>();
+ Object reader;
+ while ((reader = pool.poll()) != null) {
+ held.add(reader);
+ }
+ return held;
+ }
+
+ private static void returnToReadPool(SqliteStatsStore store, Deque<Object> held)
+ throws Exception {
+ BlockingQueue<Object> pool = readPool(store);
+ for (Object reader : held) {
+ pool.offer(reader);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static BlockingQueue<Object> readPool(SqliteStatsStore store)
+ throws Exception {
+ java.lang.reflect.Field field = SqliteStatsStore.class.getDeclaredField("_readPool");
+ field.setAccessible(true);
+ return (BlockingQueue<Object>) field.get(store);
+ }
+
+ // ---------------------------------------------------------------------------
+ // Factory helpers
+ // ---------------------------------------------------------------------------
+
+ private static SegmentStatsRow seg(String name, long crc, long docs, long sizeBytes,
+ long startMs, long endMs, boolean consuming) {
+ return new SegmentStatsRow(name, crc, docs, sizeBytes, startMs, endMs, consuming);
+ }
+
+ /// Recursively deletes a directory tree.
+ private static void deleteRecursively(Path dir)
+ throws IOException {
+ if (dir == null || !Files.exists(dir)) {
+ return;
+ }
+ try (var stream = Files.walk(dir)) {
+ stream.sorted(java.util.Comparator.reverseOrder())
+ .forEach(p -> {
+ try {
+ Files.deleteIfExists(p);
+ } catch (IOException e) {
+ // ignore
+ }
+ });
+ }
+ }
+}
diff --git a/pinot-broker/src/test/java/org/apache/pinot/broker/stats/StatsStoreContractTest.java b/pinot-broker/src/test/java/org/apache/pinot/broker/stats/StatsStoreContractTest.java
new file mode 100644
index 00000000000..08bcbdfda4c
--- /dev/null
+++ b/pinot-broker/src/test/java/org/apache/pinot/broker/stats/StatsStoreContractTest.java
@@ -0,0 +1,560 @@
+/**
+ * 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.pinot.broker.stats;
+
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.Set;
+import org.apache.pinot.query.planner.spi.stats.ColumnStatistics;
+import org.apache.pinot.query.planner.spi.stats.ColumnValueType;
+import org.apache.pinot.query.planner.spi.stats.SegmentColumnStatsRow;
+import org.apache.pinot.query.planner.spi.stats.SegmentStatsRow;
+import org.apache.pinot.query.planner.spi.stats.StatConfidence;
+import org.apache.pinot.query.planner.spi.stats.StatsAggregations;
+import org.apache.pinot.query.planner.spi.stats.StatsStore;
+import org.apache.pinot.query.planner.spi.stats.StatsStoreException;
+import org.apache.pinot.query.planner.spi.stats.TableStatistics;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.*;
+
+
+/// Behavior every [StatsStore] implementation must share.
+///
+/// The optimizer must not change with the configured store, so these cases are defined once here
+/// and run against each implementation by a subclass. Anything specific to one implementation
+/// (durability, corruption recovery) belongs in that subclass instead.
+public abstract class StatsStoreContractTest {
+ protected static final String TABLE_A = "myTable_OFFLINE";
+ protected static final String TABLE_B = "otherTable_REALTIME";
+
+ protected StatsStore _store;
+
+ /// Creates a fresh, un-initialized store for one test method.
+ protected abstract StatsStore createStore()
+ throws Exception;
+
+ /// Releases anything the concrete store needed beyond [StatsStore#close()].
+ protected void cleanUp()
+ throws Exception {
+ }
+
+ @BeforeMethod
+ public void setUp()
+ throws Exception {
+ _store = createStore();
+ _store.init();
+ }
+
+ @AfterMethod
+ public void tearDown()
+ throws Exception {
+ if (_store != null) {
+ _store.close();
+ }
+ cleanUp();
+ }
+
+ // ---------------------------------------------------------------------------
+ // Round-trip: upsert + getTableStats
+ // ---------------------------------------------------------------------------
+
+ @Test
+ public void testRoundTripTableStats()
+ throws Exception {
+ List<SegmentStatsRow> rowsA = List.of(
+ seg("seg1", 100L, 1000L, 5000L, 0L, 100L, false),
+ seg("seg2", 200L, 2000L, 8000L, 100L, 200L, false)
+ );
+ List<SegmentStatsRow> rowsB = List.of(
+ seg("seg3", 300L, 500L, 1000L, 0L, 50L, false)
+ );
+
+ _store.upsertSegmentStats(TABLE_A, rowsA);
+ _store.upsertSegmentStats(TABLE_B, rowsB);
+
+ TableStatistics statsA = _store.getTableStats(TABLE_A);
+ assertNotNull(statsA);
+ assertEquals(statsA.getRowCount(), 3000L);
+ assertEquals(statsA.getTableSizeBytes(), 13000L);
+ assertEquals(statsA.getRowCountConfidence(), StatConfidence.EXACT);
+ assertEquals(statsA.getSizeConfidence(), StatConfidence.EXACT);
+
+ TableStatistics statsB = _store.getTableStats(TABLE_B);
+ assertNotNull(statsB);
+ assertEquals(statsB.getRowCount(), 500L);
+ }
+
+ @Test
+ public void testConsumingSegmentsExcludedFromTableStats()
+ throws Exception {
+ List<SegmentStatsRow> rows = List.of(
+ seg("committed", 111L, 1000L, 5000L, 0L, 100L, false),
+ seg("consuming", 222L, 999L, 1000L, 100L, 200L, true)
+ );
+ _store.upsertSegmentStats(TABLE_A, rows);
+
+ TableStatistics stats = _store.getTableStats(TABLE_A);
+ assertNotNull(stats);
+ // Only the committed segment should be counted
+ assertEquals(stats.getRowCount(), 1000L);
+ assertEquals(stats.getTableSizeBytes(), 5000L);
+ }
+
+ @Test
+ public void testGetTableStatsNullWhenEmpty()
+ throws Exception {
+ assertNull(_store.getTableStats(TABLE_A));
+ }
+
+ // ---------------------------------------------------------------------------
+ // Upsert overwrite
+ // ---------------------------------------------------------------------------
+
+ @Test
+ public void testUpsertOverwrite()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A,
+ List.of(seg("seg1", 100L, 1000L, 5000L, 0L, 100L, false)));
+
+ // Overwrite with new values
+ _store.upsertSegmentStats(TABLE_A,
+ List.of(seg("seg1", 999L, 2500L, 9000L, 0L, 100L, false)));
+
+ Map<String, Long> crcs = _store.getSegmentCrcs(TABLE_A);
+ assertEquals(crcs.size(), 1);
+ assertEquals(crcs.get("seg1").longValue(), 999L);
+
+ TableStatistics stats = _store.getTableStats(TABLE_A);
+ assertNotNull(stats);
+ assertEquals(stats.getRowCount(), 2500L);
+ }
+
+ // ---------------------------------------------------------------------------
+ // removeSegments
+ // ---------------------------------------------------------------------------
+
+ @Test
+ public void testRemoveSegments()
+ throws Exception {
+ List<SegmentStatsRow> rows = List.of(
+ seg("seg1", 1L, 100L, 1000L, 0L, 10L, false),
+ seg("seg2", 2L, 200L, 2000L, 10L, 20L, false)
+ );
+ _store.upsertSegmentStats(TABLE_A, rows);
+ _store.upsertSegmentColumnStats(TABLE_A, List.of(
+ col("seg1", "colA", 10L, "1", "9", true, 4.0, 0.0),
+ col("seg2", "colA", 20L, "2", "8", true, 4.0, 0.0)
+ ));
+
+ _store.removeSegments(TABLE_A, List.of("seg1"));
+
+ Map<String, Long> crcs = _store.getSegmentCrcs(TABLE_A);
+ assertFalse(crcs.containsKey("seg1"));
+ assertTrue(crcs.containsKey("seg2"));
+
+ // Column stats for seg1 should also be removed
+ ColumnStatistics colStats = _store.getColumnStats(TABLE_A, "colA");
+ assertNotNull(colStats);
+ // Only seg2's stats remain: ndv=20, min=2, max=8
+ assertEquals(colStats.getNdv(), 20L);
+ }
+
+ // ---------------------------------------------------------------------------
+ // Column stats aggregation
+ // ---------------------------------------------------------------------------
+
+ @Test
+ public void testColumnStatsAggregation()
+ throws Exception {
+ // 3 segments with different stats; docs used for weighting
+ List<SegmentStatsRow> segs = List.of(
+ seg("s1", 1L, 100L, 1000L, 0L, 100L, false),
+ seg("s2", 2L, 200L, 2000L, 100L, 200L, false),
+ seg("s3", 3L, 300L, 3000L, 200L, 300L, false)
+ );
+ _store.upsertSegmentStats(TABLE_A, segs);
+
+ // ndv: 10, 20, 30 → MAX = 30
+ // min: "9", "10", "2" → numeric min = 2 (i.e. "2")
+ // max: "9", "10", "20" → numeric max = 20 (i.e. "20")
+ // minTrusted: true, false, true → AND = false
+ // avgBytes: 4.0, 8.0, 2.0 (weighted by docs 100/200/300)
+ // weighted = (4*100 + 8*200 + 2*300) / 600 = (400+1600+600)/600 = 2600/600 ≈ 4.333...
+ // nullFraction: 0.1, 0.2, 0.0 (weighted)
+ // weighted = (0.1*100 + 0.2*200 + 0.0*300) / 600 = (10+40+0)/600 = 50/600 ≈ 0.0833...
+ List<SegmentColumnStatsRow> cols = List.of(
+ col("s1", "colA", 10L, "9", "9", true, 4.0, 0.1),
+ col("s2", "colA", 20L, "10", "10", false, 8.0, 0.2),
+ col("s3", "colA", 30L, "2", "20", true, 2.0, 0.0)
+ );
+ _store.upsertSegmentColumnStats(TABLE_A, cols);
+
+ ColumnStatistics cs = _store.getColumnStats(TABLE_A, "colA");
+ assertNotNull(cs);
+
+ assertEquals(cs.getNdv(), 30L);
+ assertEquals(cs.getNdvConfidence(), StatConfidence.ESTIMATED);
+
+ // Ordered as LONG (the recorded type), and returned as the Java type that ordering implies:
+ // min of "9","10","2" is 2, max of "9","10","20" is 20.
+ assertEquals(cs.getMinValue(), 2L);
+ assertEquals(cs.getMaxValue(), 20L);
+
+ // minTrusted: s2 is false → overall false
+ assertFalse(cs.isMinTrusted());
+
+ // Weighted avgBytes ≈ 4.333
+ assertEquals(cs.getAvgBytesPerValue(), 2600.0 / 600.0, 0.001);
+
+ // Weighted nullFraction ≈ 0.0833
+ assertEquals(cs.getNullFraction(), 50.0 / 600.0, 0.001);
+ }
+
+ @Test
+ public void testNumericMinMaxOrdering()
+ throws Exception {
+ // "9" vs "10": lexically "9" > "10", numerically "9" < "10"
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("s1", 1L, 100L, 1000L, 0L, 100L, false),
+ seg("s2", 2L, 100L, 1000L, 0L, 100L, false)
+ ));
+ _store.upsertSegmentColumnStats(TABLE_A, List.of(
+ col("s1", "colA", 5L, "9", "9", true, 4.0, 0.0),
+ col("s2", "colA", 5L, "10", "10", true, 4.0, 0.0)
+ ));
+
+ ColumnStatistics cs = _store.getColumnStats(TABLE_A, "colA");
+ assertNotNull(cs);
+ // "9" vs "10" orders numerically under LONG, where text order would put "10" first.
+ assertEquals(cs.getMinValue(), 9L);
+ assertEquals(cs.getMaxValue(), 10L);
+ }
+
+ @Test
+ public void testColumnStatsNullWhenNoRows()
+ throws Exception {
+ assertNull(_store.getColumnStats(TABLE_A, "colA"));
+ }
+
+ // ---------------------------------------------------------------------------
+ // estimateRowsInTimeRange
+ // ---------------------------------------------------------------------------
+
+ @Test
+ public void testEstimateRowsFullOverlap()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("s1", 1L, 1000L, 0L, 0L, 100L, false), // [0, 100)
+ seg("s2", 2L, 2000L, 0L, 100L, 200L, false) // [100, 200)
+ ));
+
+ // Query [0, 200) → both fully inside
+ OptionalLong result = _store.estimateRowsInTimeRange(TABLE_A, 0L, 200L);
+ assertTrue(result.isPresent());
+ assertEquals(result.getAsLong(), 3000L);
+ }
+
+ @Test
+ public void testEstimateRowsPartialOverlap()
+ throws Exception {
+ // Segment [0, 100), docs=1000
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("s1", 1L, 1000L, 0L, 0L, 100L, false)
+ ));
+
+ // Query [25, 75) → 50% overlap → 500 rows
+ OptionalLong result = _store.estimateRowsInTimeRange(TABLE_A, 25L, 75L);
+ assertTrue(result.isPresent());
+ assertEquals(result.getAsLong(), 500L);
+ }
+
+ @Test
+ public void testEstimateRowsNoOverlap()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("s1", 1L, 1000L, 0L, 0L, 100L, false)
+ ));
+
+ // Query [200, 300) → no overlap
+ OptionalLong result = _store.estimateRowsInTimeRange(TABLE_A, 200L, 300L);
+ assertTrue(result.isPresent());
+ assertEquals(result.getAsLong(), 0L);
+ }
+
+ @Test
+ public void testEstimateRowsUnknownTimesSegment()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("s1", 1L, 1000L, 0L, -1L, -1L, false)
+ ));
+
+ // Unknown times → conservative, always include
+ OptionalLong result = _store.estimateRowsInTimeRange(TABLE_A, 0L, 100L);
+ assertTrue(result.isPresent());
+ assertEquals(result.getAsLong(), 1000L);
+ }
+
+ @Test
+ public void testEstimateRowsEmptyOptionalWhenNoRows()
+ throws Exception {
+ OptionalLong result = _store.estimateRowsInTimeRange(TABLE_A, 0L, 100L);
+ assertFalse(result.isPresent());
+ }
+
+ @Test
+ public void testEstimateRowsEmptyOptionalWhenOnlyConsumingSegments()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("s1", 1L, 1000L, 0L, 0L, 100L, true)
+ ));
+
+ // Consuming segments are excluded from time-range estimates; with no committed segments the
+ // result must be empty ("no stats"), never of(0) ("provably zero rows in range").
+ OptionalLong result = _store.estimateRowsInTimeRange(TABLE_A, 0L, 100L);
+ assertFalse(result.isPresent());
+ }
+
+ @Test
+ public void testEstimateRowsBoundaryAdjacency()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("s1", 1L, 1000L, 0L, 100L, 200L, false) // [100, 200)
+ ));
+
+ // Segment ending exactly at the range start → no overlap (half-open semantics)
+ OptionalLong before = _store.estimateRowsInTimeRange(TABLE_A, 200L, 300L);
+ assertTrue(before.isPresent());
+ assertEquals(before.getAsLong(), 0L);
+
+ // Segment starting exactly at the range end → no overlap
+ OptionalLong after = _store.estimateRowsInTimeRange(TABLE_A, 0L, 100L);
+ assertTrue(after.isPresent());
+ assertEquals(after.getAsLong(), 0L);
+
+ // Touching on both sides at once: range exactly equal to the segment → full overlap
+ OptionalLong exact = _store.estimateRowsInTimeRange(TABLE_A, 100L, 200L);
+ assertTrue(exact.isPresent());
+ assertEquals(exact.getAsLong(), 1000L);
+ }
+
+ // ---------------------------------------------------------------------------
+ // purgeTable / purgeAll
+ // ---------------------------------------------------------------------------
+
+ @Test
+ public void testPurgeTable()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("seg1", 1L, 100L, 1000L, 0L, 10L, false)
+ ));
+ _store.upsertSegmentStats(TABLE_B, List.of(
+ seg("seg2", 2L, 200L, 2000L, 0L, 10L, false)
+ ));
+ _store.upsertSegmentColumnStats(TABLE_A, List.of(
+ col("seg1", "colA", 5L, "1", "9", true, 4.0, 0.0)
+ ));
+
+ _store.purgeTable(TABLE_A);
+
+ assertNull(_store.getTableStats(TABLE_A));
+ assertNull(_store.getColumnStats(TABLE_A, "colA"));
+ // TABLE_B should still be there
+ assertNotNull(_store.getTableStats(TABLE_B));
+ }
+
+ @Test
+ public void testPurgeAll()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(
+ seg("seg1", 1L, 100L, 1000L, 0L, 10L, false)
+ ));
+ _store.upsertSegmentStats(TABLE_B, List.of(
+ seg("seg2", 2L, 200L, 2000L, 0L, 10L, false)
+ ));
+
+ _store.purgeAll();
+
+ assertNull(_store.getTableStats(TABLE_A));
+ assertNull(_store.getTableStats(TABLE_B));
+ }
+
+ // ---------------------------------------------------------------------------
+ // Consuming segments
+ // ---------------------------------------------------------------------------
+
+ /// Drives whether realtime row counts are trusted (see `LogicalTableStatsResolver`), so both
+ /// stores must agree on it.
+ @Test
+ public void testHasConsumingSegments()
+ throws Exception {
+ assertFalse(_store.hasConsumingSegments(TABLE_A), "Unknown table has no consuming segments");
+
+ _store.upsertSegmentStats(TABLE_A, List.of(seg("committed", 1L, 100L, 1000L, 0L, 10L, false)));
+ assertFalse(_store.hasConsumingSegments(TABLE_A));
+
+ _store.upsertSegmentStats(TABLE_A, List.of(seg("consuming", 2L, 50L, 500L, 10L, 20L, true)));
+ assertTrue(_store.hasConsumingSegments(TABLE_A));
+
+ _store.removeSegments(TABLE_A, List.of("consuming"));
+ assertFalse(_store.hasConsumingSegments(TABLE_A), "Removing the consuming segment clears the flag");
+ }
+
+ /// A table with nothing but consuming segments has no trustworthy aggregate: reads must report
+ /// "no statistics" rather than a zero, and the segment must still be reconcilable by crc.
+ @Test
+ public void testOnlyConsumingSegmentsYieldNoTableOrColumnStats()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(seg("consuming", 7L, 999L, 500L, 0L, 10L, true)));
+ _store.upsertSegmentColumnStats(TABLE_A,
+ List.of(col("consuming", "colA", 5L, "1", "9", true, 4.0, 0.0)));
+
+ assertNull(_store.getTableStats(TABLE_A), "Consuming-only table has no committed rows to aggregate");
+ assertNull(_store.getColumnStats(TABLE_A, "colA"), "A column row on a consuming segment does not count");
+ assertTrue(_store.getSegmentCrcs(TABLE_A).containsKey("consuming"), "crcs still cover consuming segments");
+ }
+
+ /// A column row whose segment row is gone contributes nothing: it supplies the doc count used for
+ /// weighting, so counting it would weight by a stale or absent value.
+ @Test
+ public void testColumnStatsIgnoreRowsWithoutASegment()
+ throws Exception {
+ _store.upsertSegmentColumnStats(TABLE_A,
+ List.of(col("ghost", "colA", 5L, "1", "9", true, 4.0, 0.0)));
+
+ assertNull(_store.getColumnStats(TABLE_A, "colA"));
+ }
+
+ // ---------------------------------------------------------------------------
+ // Shared aggregation semantics
+ // ---------------------------------------------------------------------------
+
+ /// Pins the ordering rules both stores rely on. The type has to be recorded with the values:
+ /// guessing "numeric if it parses" orders a STRING column numerically and rounds a LONG past
+ /// 2^53 — inwards, which would exclude rows that exist.
+ @Test
+ public void testValueOrderingIsDrivenByTheRecordedType() {
+ assertTrue(ColumnValueType.LONG.compare("9", "10") < 0, "LONG orders numerically");
+ assertTrue(ColumnValueType.STRING.compare("9", "10") > 0, "STRING orders lexically");
+
+ // 2^53 + 1 and 2^53 + 3 are indistinguishable as doubles; as LONG they are not.
+ assertTrue(ColumnValueType.LONG.compare("9007199254740993", "9007199254740995") < 0,
+ "LONG must not lose precision past 2^53");
+ assertEquals(ColumnValueType.LONG.toComparable("9007199254740993"), 9007199254740993L);
+ assertEquals(ColumnValueType.BIG_DECIMAL.toComparable("1.5"), new java.math.BigDecimal("1.5"));
+ assertEquals(ColumnValueType.STRING.toComparable("abc"), "abc");
+ assertNull(ColumnValueType.STRING.toComparable(null));
+
+ assertEquals(StatsAggregations.minOf(null, "5", ColumnValueType.LONG), "5", "null means unknown");
+ assertEquals(StatsAggregations.maxOf("5", null, ColumnValueType.LONG), "5");
+ assertNull(StatsAggregations.minOf(null, null, ColumnValueType.LONG));
+ }
+
+ /// Without a recorded type there is no defined ordering, so the bounds must be reported
+ /// untrusted rather than ordered on a guess — a consumer may prune on trusted bounds.
+ @Test
+ public void testUnknownValueTypeYieldsUntrustedBounds()
+ throws Exception {
+ _store.upsertSegmentStats(TABLE_A, List.of(seg("s1", 1L, 100L, 1000L, 0L, 10L, false)));
+ _store.upsertSegmentColumnStats(TABLE_A,
+ List.of(new SegmentColumnStatsRow("s1", "colA", 5L, "9", "10", true, 4.0, 0.0, null)));
+
+ ColumnStatistics stats = _store.getColumnStats(TABLE_A, "colA");
+ assertNotNull(stats);
+ assertFalse(stats.isMinTrusted(), "Bounds with no recorded ordering must not be trusted");
+ }
+
+ /// Pins the per-segment overlap rules, including the cases a query never makes obvious.
+ @Test
+ public void testOverlapSemantics() {
+ assertEquals(StatsAggregations.overlapRows(100L, -1L, -1L, 0L, 10L), 100L, "Unknown times count in full");
+ assertEquals(StatsAggregations.overlapRows(100L, 10L, 20L, 0L, 100L), 100L, "Full containment");
+ assertEquals(StatsAggregations.overlapRows(100L, 100L, 200L, 0L, 100L), 0L, "Half-open upper bound excludes");
+ assertEquals(StatsAggregations.overlapRows(1000L, 0L, 100L, 25L, 75L), 500L, "Partial overlap interpolates");
+ assertEquals(StatsAggregations.overlapRows(100L, 50L, 50L, 0L, 100L), 100L, "Zero-length segment in range");
+ assertEquals(StatsAggregations.overlapRows(100L, 50L, 50L, 60L, 100L), 0L, "Zero-length segment out of range");
+ }
+
+ // ---------------------------------------------------------------------------
+ // Enumerating stored tables
+ // ---------------------------------------------------------------------------
+
+ /// getTables() feeds a destructive purge, so both stores must agree on what "holds statistics"
+ /// means -- including that a table whose segments are all gone is no longer reported.
+ @Test
+ public void testGetTablesReflectsStoredRows()
+ throws Exception {
+ assertTrue(_store.getTables().isEmpty(), "A fresh store holds nothing");
+
+ _store.upsertSegmentStats(TABLE_A, List.of(seg("s1", 1L, 100L, 1000L, 0L, 10L, false)));
+ _store.upsertSegmentStats(TABLE_B, List.of(seg("s2", 2L, 200L, 2000L, 0L, 10L, false)));
+ assertEquals(_store.getTables(), Set.of(TABLE_A, TABLE_B));
+
+ _store.removeSegments(TABLE_A, List.of("s1"));
+ assertEquals(_store.getTables(), Set.of(TABLE_B), "A table with no rows left is not reported");
+
+ _store.purgeTable(TABLE_B);
+ assertTrue(_store.getTables().isEmpty());
+ }
+
+ // ---------------------------------------------------------------------------
+ // Closed store
+ // ---------------------------------------------------------------------------
+
+ @Test
+ public void testReadsAfterCloseAreRejected()
+ throws Exception {
+ _store.close();
+ assertThrows(StatsStoreException.class, () -> _store.getTableStats(TABLE_A));
+ assertThrows(StatsStoreException.class, () -> _store.getTables());
+ _store = null; // already closed; tearDown must not close it twice
+ }
+
+ // ---------------------------------------------------------------------------
+ // getSegmentCrcs
+ // ---------------------------------------------------------------------------
+
+ @Test
+ public void testGetSegmentCrcsEmptyWhenNoData()
+ throws Exception {
+ Map<String, Long> crcs = _store.getSegmentCrcs(TABLE_A);
+ assertNotNull(crcs);
+ assertTrue(crcs.isEmpty());
+ }
+
+ // ---------------------------------------------------------------------------
+ // Factory helpers
+ // ---------------------------------------------------------------------------
+
+ private static SegmentStatsRow seg(String name, long crc, long docs, long sizeBytes,
+ long startMs, long endMs, boolean consuming) {
+ return new SegmentStatsRow(name, crc, docs, sizeBytes, startMs, endMs, consuming);
+ }
+
+ /// Existing cases use numeric min/max, so they record LONG; the untyped case is exercised
+ /// explicitly by testUnknownValueTypeYieldsUntrustedBounds.
+ private static SegmentColumnStatsRow col(String segName, String colName, long ndv,
+ String minVal, String maxVal, boolean minTrusted, double avgBytes, double nullFrac) {
+ return new SegmentColumnStatsRow(segName, colName, ndv, minVal, maxVal, minTrusted, avgBytes,
+ nullFrac, ColumnValueType.LONG);
+ }
+}
diff --git a/pom.xml b/pom.xml
index 10eedc3643f..6b730e954a7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -362,6 +362,7 @@
<equalsverifier.version>3.19.4</equalsverifier.version>
<testcontainers.version>2.0.5</testcontainers.version>
<h2.version>2.4.240</h2.version>
+ <sqlite-jdbc.version>3.46.1.3</sqlite-jdbc.version>
<jnr-posix.version>3.2.2</jnr-posix.version>
<scalatest.version>3.2.20</scalatest.version>
<assertj.version>3.27.7</assertj.version>
@@ -2084,6 +2085,13 @@
<scope>test</scope>
</dependency>
+ <!-- SQLite JDBC driver (broker-local stats store) -->
+ <dependency>
+ <groupId>org.xerial</groupId>
+ <artifactId>sqlite-jdbc</artifactId>
+ <version>${sqlite-jdbc.version}</version>
+ </dependency>
+
<!-- assertj bom -->
<dependency>
<groupId>org.assertj</groupId>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
(pinot) 01/02: Add statistics contracts for cost-based optimization
Posted by go...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
gortiz pushed a commit to branch cbo-1-stats-contracts-and-stores
in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 3a1b835cc591e554b6ac2248a0d4df7ba252d7ab
Author: Gonzalo Ortiz <19...@users.noreply.github.com>
AuthorDate: Thu Aug 27 20:02:17 2026 +0200
Add statistics contracts for cost-based optimization
Defines what a Pinot statistic is, before anything produces or consumes one:
TableStatistics and ColumnStatistics as the values, StatConfidence as the trust
tier attached to each one, StatsStore as broker-local persistence over
per-segment rows, and ColumnStatsSource as where per-column statistics are
fetched from.
Confidence is per statistic rather than per table, so a source that can only
estimate some values does not have to devalue the rest. Consumers are expected
to treat a low-confidence statistic as absent instead of trusting it, which is
what keeps table types with biased raw counts (upsert, dedup, consuming
segments) on today's behavior rather than producing confidently wrong plans.
StatsAggregations carries the rollup semantics every store must share, so two
implementations cannot disagree about what the same stored rows mean: time
overlap interpolation, document-weighted averages that exclude the "unknown"
sentinel rather than averaging it in as a measurement, and min/max folded under
the ordering the column actually has.
That ordering is recorded per row as a ColumnValueType rather than guessed from
the text, because guessing is wrong in both directions: a string column holding
"9" and "10" orders lexically in Pinot but would compare numerically, and a long
beyond 2^53 loses digits as a double -- in the direction that narrows the range,
which would exclude rows that exist. When the ordering is unknown or segments
disagree about it, both bounds are reported as absent: a bound folded under two
different orderings is neither a true minimum nor a true maximum, and there is
no honest way to describe it as merely untrusted.
The store contract includes getTables(), so a caller can find tables it no
longer serves. Per-table cleanup is otherwise driven by an event -- a routing
entry being removed -- which a broker cannot observe for a table dropped while
it was down.
---
CLAUDE.md | 6 +
.../query/planner/spi/stats/ColumnStatistics.java | 180 ++++++++++++++++++++
.../spi/stats/ColumnStatsFetchException.java | 47 ++++++
.../query/planner/spi/stats/ColumnStatsSource.java | 56 ++++++
.../query/planner/spi/stats/ColumnValueType.java | 107 ++++++++++++
.../planner/spi/stats/SegmentColumnStatsRow.java | 113 +++++++++++++
.../query/planner/spi/stats/SegmentStatsRow.java | 34 ++++
.../query/planner/spi/stats/StatConfidence.java | 47 ++++++
.../query/planner/spi/stats/StatsAggregations.java | 187 +++++++++++++++++++++
.../pinot/query/planner/spi/stats/StatsStore.java | 153 +++++++++++++++++
.../planner/spi/stats/StatsStoreException.java | 44 +++++
.../planner/spi/stats/StatsStoreProvider.java | 45 +++++
.../query/planner/spi/stats/TableStatistics.java | 116 +++++++++++++
.../planner/spi/stats/ColumnValueTypeTest.java | 106 ++++++++++++
.../planner/spi/stats/StatsAggregationsTest.java | 148 ++++++++++++++++
15 files changed, 1389 insertions(+)
diff --git a/CLAUDE.md b/CLAUDE.md
index 046bd537a09..9eae2dafd4f 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -111,6 +111,12 @@ Apache Pinot is a real-time distributed OLAP datastore for low-latency analytics
- Keep Apache 2.0 license headers on all new source files.
- Preserve backward compatibility across mixed-version broker/server/controller.
- Prefer imports over fully qualified class names (e.g., use `import com.foo.Bar` and refer to `Bar`, not `com.foo.Bar` inline).
+- Prefer records for immutable value carriers, especially in new APIs. A record states the
+ intent (data, not behaviour), gives correct `equals`/`hashCode`/`toString` for free, and cannot
+ drift out of sync with its fields. Use a class only when the type genuinely needs mutability,
+ inheritance, or a non-trivial identity contract. Note that a record does not excuse a dangerous
+ constructor: when a record has several components of the same type in a row, add a builder so
+ call sites name what they pass — a record and a builder compose fine.
- Prefer `List.of()`, `Set.of()`, and `Map.of()` for non-null immutable collection literals. Checkstyle blocks
`Collections.emptyList()`, `Collections.emptySet()`, and `Collections.emptyMap()`; use `List.of()`, `Set.of()`, and
`Map.of()` instead. Do not add blanket bans for `Collections.singleton*`; use them only when an element/key/value
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnStatistics.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnStatistics.java
new file mode 100644
index 00000000000..6377258252f
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnStatistics.java
@@ -0,0 +1,180 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+import java.util.Objects;
+import javax.annotation.Nullable;
+
+
+/// Immutable per-column statistics for a table, used by the cost-based query planner.
+///
+/// Instances are created via [#builder()]. Unknown numeric fields are represented
+/// by `-1`. Min/max values may be `null` when unknown.
+///
+/// Thread-safety: immutable; safe for concurrent access.
+public class ColumnStatistics {
+ private final String _columnName;
+ private final long _ndv;
+ private final StatConfidence _ndvConfidence;
+ @Nullable
+ private final Comparable<?> _minValue;
+ @Nullable
+ private final Comparable<?> _maxValue;
+ private final boolean _minTrusted;
+ private final double _avgBytesPerValue;
+ private final double _nullFraction;
+
+ private ColumnStatistics(Builder builder) {
+ _columnName = builder._columnName;
+ _ndv = builder._ndv;
+ _ndvConfidence = builder._ndvConfidence;
+ _minValue = builder._minValue;
+ _maxValue = builder._maxValue;
+ _minTrusted = builder._minTrusted;
+ _avgBytesPerValue = builder._avgBytesPerValue;
+ _nullFraction = builder._nullFraction;
+ }
+
+ /// Returns a new [Builder] for constructing [ColumnStatistics] instances.
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /// Returns the name of the column these statistics describe.
+ public String getColumnName() {
+ return _columnName;
+ }
+
+ /// Returns the estimated number of distinct values (NDV) for the column, or `-1` if unknown.
+ public long getNdv() {
+ return _ndv;
+ }
+
+ /// Returns the confidence level of the [#getNdv()] value.
+ public StatConfidence getNdvConfidence() {
+ return _ndvConfidence;
+ }
+
+ /// Returns the minimum observed value for the column, or `null` if unknown.
+ @Nullable
+ public Comparable<?> getMinValue() {
+ return _minValue;
+ }
+
+ /// Returns the maximum observed value for the column, or `null` if unknown.
+ @Nullable
+ public Comparable<?> getMaxValue() {
+ return _maxValue;
+ }
+
+ /// Returns `false` when the minimum value is polluted by the numeric null-sentinel default
+ /// (i.e. the column is nullable and its stored min equals the type's minimum representable value).
+ /// When `false`, estimation may assume the range `[-max..max]`, and segment pruning
+ /// must never use this bound.
+ public boolean isMinTrusted() {
+ return _minTrusted;
+ }
+
+ /// Returns the average number of bytes per stored value for the column, or `-1` if unknown.
+ public double getAvgBytesPerValue() {
+ return _avgBytesPerValue;
+ }
+
+ /// Returns the fraction of rows where the column value is null (in the range `[0.0, 1.0]`),
+ /// or `-1` if unknown.
+ public double getNullFraction() {
+ return _nullFraction;
+ }
+
+ /// Builder for [ColumnStatistics].
+ ///
+ /// [#columnName(String)] is required; every other field is optional. Default values: string
+ /// fields default to `null`, numeric fields to `-1` (unknown), confidence fields to
+ /// [StatConfidence#UNKNOWN], and `minTrusted` defaults to `true`.
+ ///
+ /// Thread-safety: not thread-safe; use from a single thread.
+ public static class Builder {
+ private String _columnName;
+ private long _ndv = -1;
+ private StatConfidence _ndvConfidence = StatConfidence.UNKNOWN;
+ @Nullable
+ private Comparable<?> _minValue;
+ @Nullable
+ private Comparable<?> _maxValue;
+ private boolean _minTrusted = true;
+ private double _avgBytesPerValue = -1;
+ private double _nullFraction = -1;
+
+ private Builder() {
+ }
+
+ /// Sets the column name.
+ public Builder columnName(String columnName) {
+ _columnName = columnName;
+ return this;
+ }
+
+ /// Sets the number of distinct values (NDV) and its confidence level.
+ public Builder ndv(long ndv, StatConfidence confidence) {
+ _ndv = ndv;
+ _ndvConfidence = confidence;
+ return this;
+ }
+
+ /// Sets the minimum observed value.
+ public Builder minValue(@Nullable Comparable<?> minValue) {
+ _minValue = minValue;
+ return this;
+ }
+
+ /// Sets the maximum observed value.
+ public Builder maxValue(@Nullable Comparable<?> maxValue) {
+ _maxValue = maxValue;
+ return this;
+ }
+
+ /// Sets whether the minimum value is trustworthy (i.e. not polluted by a null-sentinel
+ /// default).
+ public Builder minTrusted(boolean minTrusted) {
+ _minTrusted = minTrusted;
+ return this;
+ }
+
+ /// Sets the average number of bytes per stored value.
+ public Builder avgBytesPerValue(double avgBytesPerValue) {
+ _avgBytesPerValue = avgBytesPerValue;
+ return this;
+ }
+
+ /// Sets the null fraction (proportion of null values in `[0.0, 1.0]`).
+ public Builder nullFraction(double nullFraction) {
+ _nullFraction = nullFraction;
+ return this;
+ }
+
+ /// Builds the immutable [ColumnStatistics] instance.
+ ///
+ /// @throws NullPointerException if no column name was set; [#getColumnName()] is declared
+ /// non-null, so a nameless instance would hand callers a null they cannot see coming.
+ public ColumnStatistics build() {
+ Objects.requireNonNull(_columnName, "columnName must be set");
+ return new ColumnStatistics(this);
+ }
+ }
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnStatsFetchException.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnStatsFetchException.java
new file mode 100644
index 00000000000..ba47ae0ccf6
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnStatsFetchException.java
@@ -0,0 +1,47 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+
+/// Checked exception thrown by [ColumnStatsSource] when column statistics cannot be fetched and no
+/// partial result can be returned.
+///
+/// Throwing is the last resort: a source that obtained statistics for some segments should return
+/// them and omit the rest, since a partial result still improves estimates. Callers must treat this
+/// as "no new statistics this round" and degrade to whatever is already stored — collection failures
+/// must never fail a query.
+///
+/// Thread-safety: exception objects are not shared; no concurrency requirements.
+public class ColumnStatsFetchException extends Exception {
+
+ /// Constructs a new [ColumnStatsFetchException] with the given message.
+ ///
+ /// @param message description of the error
+ public ColumnStatsFetchException(String message) {
+ super(message);
+ }
+
+ /// Constructs a new [ColumnStatsFetchException] with the given message and cause.
+ ///
+ /// @param message description of the error
+ /// @param cause the underlying cause
+ public ColumnStatsFetchException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnStatsSource.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnStatsSource.java
new file mode 100644
index 00000000000..056158e9981
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnStatsSource.java
@@ -0,0 +1,56 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+
+/// Source of per-segment column statistics fetched on behalf of the broker: the acquisition side
+/// of the statistics subsystem, paired with [StatsStore] (where statistics are kept) and
+/// [PinotStatisticsProvider] (how the planner reads them).
+///
+/// This is an extension point on purpose. A server fan-out, a push feed and a vendor metadata
+/// service have very different cost profiles, so the choice of source must not be baked into the
+/// collection logic.
+///
+/// Implementations own their own bounding strategy (rate limits, jitter, debounce) and are
+/// responsible for not overwhelming downstream services.
+///
+/// The result is keyed by segment name; segments for which statistics could not be obtained
+/// may be absent from the returned map.
+///
+/// Thread-safety: implementations must be thread-safe.
+public interface ColumnStatsSource {
+
+ /// Fetches per-column statistics for the specified segments of the given table.
+ ///
+ /// Segments for which statistics are unavailable may be absent from the result map.
+ /// Implementations may return a partial result on partial failure.
+ ///
+ /// @param tableNameWithType fully-qualified table name including type suffix
+ /// @param segmentNames names of the segments for which statistics are requested
+ /// @return map from segment name to a list of per-column statistics rows; missing segments are
+ /// absent from the map
+ /// @throws ColumnStatsFetchException if fetching fails and no partial result can be returned
+ Map<String, List<SegmentColumnStatsRow>> fetchColumnStats(String tableNameWithType,
+ Set<String> segmentNames)
+ throws ColumnStatsFetchException;
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnValueType.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnValueType.java
new file mode 100644
index 00000000000..b8f58dbc515
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/ColumnValueType.java
@@ -0,0 +1,107 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+import java.math.BigDecimal;
+import javax.annotation.Nullable;
+
+
+/// How a column's min/max values, which are stored as text, must be ordered and deserialized.
+///
+/// The type has to be recorded with the values because it cannot be recovered from them. Guessing
+/// "numeric if it parses" is wrong in both directions: a STRING column holding `"9"` and `"10"`
+/// orders lexically in Pinot but would compare numerically, and a LONG beyond 2^53 loses precision
+/// as a `double` — in the direction that narrows the range, which would exclude rows that exist.
+///
+/// This enum deliberately mirrors only the ordering classes the statistics layer needs rather than
+/// Pinot's full `DataType`: this module does not depend on `pinot-spi` at compile scope, and the
+/// statistics layer only ever needs to know how to order a value, not how it is encoded. Producers
+/// map their column type onto it.
+public enum ColumnValueType {
+ /// Exact integral ordering; parsed as [Long] so values beyond 2^53 keep every digit.
+ LONG,
+ /// Floating-point ordering.
+ DOUBLE,
+ /// Arbitrary-precision numeric ordering.
+ BIG_DECIMAL,
+ /// Lexical ordering, matching how Pinot orders string columns.
+ STRING;
+
+ /// Resolves a persisted type name, returning `null` for any name this build does not know.
+ ///
+ /// Deliberately not [#valueOf(String)]: the name comes from a store that can outlive the process
+ /// that wrote it, so a broker reading a file written by a newer build -- an aborted rolling
+ /// upgrade, say -- would otherwise throw [IllegalArgumentException] out of the query-planning
+ /// path. `null` is already the documented "ordering unknown" value, which degrades to untrusted
+ /// bounds, so an unrecognized name costs precision rather than the query.
+ @Nullable
+ public static ColumnValueType fromName(@Nullable String name) {
+ if (name == null) {
+ return null;
+ }
+ for (ColumnValueType type : values()) {
+ if (type.name().equals(name)) {
+ return type;
+ }
+ }
+ return null;
+ }
+
+ /// Orders two stored values of this type. Values that cannot be parsed as this type fall back to
+ /// lexical order, so a malformed row degrades rather than throwing on the planning path.
+ public int compare(String a, String b) {
+ try {
+ switch (this) {
+ case LONG:
+ return Long.compare(Long.parseLong(a), Long.parseLong(b));
+ case DOUBLE:
+ return Double.compare(Double.parseDouble(a), Double.parseDouble(b));
+ case BIG_DECIMAL:
+ return new BigDecimal(a).compareTo(new BigDecimal(b));
+ default:
+ return a.compareTo(b);
+ }
+ } catch (NumberFormatException e) {
+ return a.compareTo(b);
+ }
+ }
+
+ /// Deserializes a stored value into the [Comparable] a consumer expects for this type, or the
+ /// raw [String] when it cannot be parsed.
+ @Nullable
+ public Comparable<?> toComparable(@Nullable String value) {
+ if (value == null) {
+ return null;
+ }
+ try {
+ switch (this) {
+ case LONG:
+ return Long.parseLong(value);
+ case DOUBLE:
+ return Double.parseDouble(value);
+ case BIG_DECIMAL:
+ return new BigDecimal(value);
+ default:
+ return value;
+ }
+ } catch (NumberFormatException e) {
+ return value;
+ }
+ }
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/SegmentColumnStatsRow.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/SegmentColumnStatsRow.java
new file mode 100644
index 00000000000..5260d2f2135
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/SegmentColumnStatsRow.java
@@ -0,0 +1,113 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+import javax.annotation.Nullable;
+
+/// Per-column statistics for a single segment, as persisted in the broker-local [StatsStore].
+///
+/// Min/max values are serialized as strings so heterogeneous column types share one schema; the
+/// ordering to compare them under is carried separately by `valueType`, because it cannot be
+/// recovered from the text. Unknown numeric fields are represented by `-1`.
+///
+/// Prefer [#builder()] over the canonical constructor: the components include two adjacent
+/// strings (min/max) and two adjacent doubles (avg bytes / null fraction), so a transposition
+/// compiles silently and corrupts statistics.
+///
+/// @param segmentName the segment name
+/// @param columnName the column name
+/// @param ndv number of distinct values, or `-1` if unknown
+/// @param minValue minimum value as text, or `null` if unknown
+/// @param maxValue maximum value as text, or `null` if unknown
+/// @param minTrusted `false` when the minimum may be polluted by a null sentinel
+/// @param avgBytesPerValue average encoded size per value, or `-1` if unknown
+/// @param nullFraction fraction of null values in `[0, 1]`, or `-1` if unknown
+/// @param valueType how min/max must be ordered, or `null` if unknown
+public record SegmentColumnStatsRow(String segmentName, String columnName, long ndv, @Nullable String minValue,
+ @Nullable String maxValue, boolean minTrusted, double avgBytesPerValue,
+ double nullFraction, @Nullable ColumnValueType valueType) {
+
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /// Names each component at the call site, so no pair of same-typed fields can be swapped
+ /// silently. Unset numeric fields default to the `-1` unknown sentinel.
+ ///
+ /// Thread-safety: not thread-safe; use from a single thread.
+ public static final class Builder {
+ private String _segmentName;
+ private String _columnName;
+ private long _ndv = -1;
+ @Nullable
+ private String _minValue;
+ @Nullable
+ private String _maxValue;
+ private boolean _minTrusted = true;
+ private double _avgBytesPerValue = -1;
+ private double _nullFraction = -1;
+ @Nullable
+ private ColumnValueType _valueType;
+
+ public Builder segmentName(String segmentName) {
+ _segmentName = segmentName;
+ return this;
+ }
+
+ public Builder columnName(String columnName) {
+ _columnName = columnName;
+ return this;
+ }
+
+ public Builder ndv(long ndv) {
+ _ndv = ndv;
+ return this;
+ }
+
+ /// Bounds and the ordering they must be compared under, set together because a bound whose
+ /// ordering is unknown cannot be used.
+ public Builder bounds(@Nullable String minValue, @Nullable String maxValue,
+ @Nullable ColumnValueType valueType) {
+ _minValue = minValue;
+ _maxValue = maxValue;
+ _valueType = valueType;
+ return this;
+ }
+
+ public Builder minTrusted(boolean minTrusted) {
+ _minTrusted = minTrusted;
+ return this;
+ }
+
+ public Builder avgBytesPerValue(double avgBytesPerValue) {
+ _avgBytesPerValue = avgBytesPerValue;
+ return this;
+ }
+
+ public Builder nullFraction(double nullFraction) {
+ _nullFraction = nullFraction;
+ return this;
+ }
+
+ public SegmentColumnStatsRow build() {
+ return new SegmentColumnStatsRow(_segmentName, _columnName, _ndv, _minValue, _maxValue, _minTrusted,
+ _avgBytesPerValue, _nullFraction, _valueType);
+ }
+ }
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/SegmentStatsRow.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/SegmentStatsRow.java
new file mode 100644
index 00000000000..6d371844e30
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/SegmentStatsRow.java
@@ -0,0 +1,34 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+/// Aggregate statistics for a single segment, as persisted in the broker-local [StatsStore].
+///
+/// Unknown numeric fields are represented by `-1`.
+///
+/// @param segmentName the segment name
+/// @param crc CRC checksum of the segment, or `-1` if unknown
+/// @param totalDocs total documents in the segment, or `-1` if unknown
+/// @param sizeBytes on-disk size in bytes, or `-1` if unknown
+/// @param startTimeMs start of the segment's time range in epoch millis, or `-1` if unknown
+/// @param endTimeMs end of the segment's time range in epoch millis, or `-1` if unknown
+/// @param consuming `true` for a consuming (REALTIME IN_PROGRESS) segment
+public record SegmentStatsRow(String segmentName, long crc, long totalDocs, long sizeBytes, long startTimeMs,
+ long endTimeMs, boolean consuming) {
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatConfidence.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatConfidence.java
new file mode 100644
index 00000000000..60dd0c45942
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatConfidence.java
@@ -0,0 +1,47 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+
+/// Indicates how trustworthy a statistics value is for cost-based query planning.
+///
+/// Callers should treat [#LOW] statistics the same as [#UNKNOWN] for
+/// cost-based decisions because LOW values are known to be systematically biased.
+///
+/// This enum is append-only — new confidence levels may be added without breaking
+/// code compiled against an older version. Existing constants must never be reordered
+/// or removed.
+///
+/// Thread-safety: enum constants are inherently thread-safe.
+public enum StatConfidence {
+ /// Derived from authoritative metadata (e.g. sum of committed segments' totalDocs for an
+ /// OFFLINE table).
+ EXACT,
+
+ /// Derived via approximation (e.g. clamped NDV merge, interpolated time ranges).
+ ESTIMATED,
+
+ /// Known to be systematically biased (e.g. upsert tables where physical doc count over-counts
+ /// logical rows; tables with consuming segments). Planner must treat LOW like absent stats for
+ /// cost-based decisions.
+ LOW,
+
+ /// No information available.
+ UNKNOWN
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsAggregations.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsAggregations.java
new file mode 100644
index 00000000000..d204c4b9c4f
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsAggregations.java
@@ -0,0 +1,187 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+import javax.annotation.Nullable;
+
+
+/// Aggregation semantics shared by every [StatsStore] implementation.
+///
+/// These rules define what the stored per-segment rows *mean*, so they must not be re-derived per
+/// implementation: two stores that disagree here would make the optimizer behave differently
+/// depending on which store an operator configured. They live beside the contract, and are public,
+/// so an implementation outside this package can reuse them rather than reinventing them.
+///
+/// Thread-safety: stateless; all methods are pure.
+public final class StatsAggregations {
+
+ private StatsAggregations() {
+ }
+
+ /// Returns how many of a segment's `docs` fall in the half-open query range
+ /// `[startMs, endMs)`, given the segment's own inclusive time range `[segStart, segEnd]`.
+ ///
+ /// - Unknown segment times (the `-1` sentinel on either bound) count in full: the segment cannot
+ /// be excluded, and over-counting is the conservative direction for an estimate.
+ /// - No overlap contributes 0.
+ /// - Full containment contributes every doc.
+ /// - Partial overlap is interpolated linearly over the segment's duration.
+ /// - A zero-length segment counts in full when its single point lies in range.
+ public static long overlapRows(long docs, long segStart, long segEnd, long startMs, long endMs) {
+ if (segStart == -1 || segEnd == -1) {
+ return docs;
+ }
+ if (segEnd <= startMs || segStart >= endMs) {
+ return 0;
+ }
+ if (segStart >= startMs && segEnd <= endMs) {
+ return docs;
+ }
+ long segDuration = segEnd - segStart;
+ if (segDuration <= 0) {
+ return segStart >= startMs && segStart < endMs ? docs : 0;
+ }
+ long overlapStart = Math.max(startMs, segStart);
+ long overlapEnd = Math.min(endMs, segEnd);
+ double fraction = (double) (overlapEnd - overlapStart) / segDuration;
+ return Math.round(docs * fraction);
+ }
+
+ /// Accumulates per-segment column rows into a single [ColumnStatistics], applying the rules that
+ /// define what those stored rows mean.
+ ///
+ /// Every store must fold rows through this, so that the estimate a query gets cannot depend on
+ /// which store an operator configured.
+ ///
+ /// Thread-safety: not thread-safe; use one accumulator per aggregation.
+ public static final class ColumnStatsAccumulator {
+ private long _maxNdv = -1;
+ private boolean _anyUntrustedMin;
+ private long _totalDocs;
+ private double _weightedAvgBytes;
+ /// Documents behind [#_weightedAvgBytes]. Tracked separately from [#_totalDocs] because rows
+ /// carrying the "unknown" sentinel contribute no weight, so dividing by the full document
+ /// count would understate the average.
+ private long _avgBytesDocs;
+ private double _weightedNullFraction;
+ /// Documents behind [#_weightedNullFraction]; see [#_avgBytesDocs].
+ private long _nullFractionDocs;
+ @Nullable
+ private String _min;
+ @Nullable
+ private String _max;
+ @Nullable
+ private ColumnValueType _valueType;
+ private boolean _typeConflict;
+ private boolean _empty = true;
+
+ /// Adds one segment's row for this column, weighted by that segment's document count.
+ public void add(long segmentDocs, SegmentColumnStatsRow row) {
+ _empty = false;
+ _maxNdv = Math.max(_maxNdv, row.ndv());
+ if (!row.minTrusted()) {
+ _anyUntrustedMin = true;
+ }
+ _totalDocs += segmentDocs;
+ // Both fields reserve a negative value for "unknown" (see SegmentColumnStatsRow). Weighting
+ // that sentinel in as if it were a measurement produces a nonsense average -- a mix of an
+ // unknown row and a known 0.2 null fraction would yield a negative fraction, which is
+ // neither a legal fraction nor the sentinel a consumer tests for.
+ double avgBytes = row.avgBytesPerValue();
+ if (avgBytes >= 0) {
+ _weightedAvgBytes += avgBytes * segmentDocs;
+ _avgBytesDocs += segmentDocs;
+ }
+ double nullFraction = row.nullFraction();
+ if (nullFraction >= 0) {
+ _weightedNullFraction += nullFraction * segmentDocs;
+ _nullFractionDocs += segmentDocs;
+ }
+
+ ColumnValueType rowType = row.valueType();
+ if (rowType == null) {
+ // No recorded ordering: the text cannot tell us how to compare, so stop trusting bounds
+ // rather than guessing one.
+ _typeConflict = true;
+ } else if (_valueType == null) {
+ _valueType = rowType;
+ } else if (_valueType != rowType) {
+ // Segments disagreeing about a column's type means one of them is stale; ordering across
+ // them is undefined.
+ _typeConflict = true;
+ }
+
+ // Once the ordering is in doubt the bounds are discarded wholesale by build(), so there is
+ // nothing to gain by folding further rows into them under a guessed ordering.
+ if (!_typeConflict && _valueType != null) {
+ _min = minOf(_min, row.minValue(), _valueType);
+ _max = maxOf(_max, row.maxValue(), _valueType);
+ }
+ }
+
+ /// Returns `true` when no row was added, in which case the caller reports "no statistics"
+ /// rather than an empty aggregate.
+ public boolean isEmpty() {
+ return _empty;
+ }
+
+ public ColumnStatistics build(String columnName) {
+ ColumnValueType effectiveType = _typeConflict || _valueType == null ? null : _valueType;
+ return ColumnStatistics.builder()
+ .columnName(columnName)
+ // Bounds folded without a known ordering are neither the true minimum nor the true
+ // maximum under any ordering, so they are reported as absent rather than as untrusted:
+ // isMinTrusted says nothing about the maximum, and a consumer following its documented
+ // remedy would build a range out of an equally unreliable bound.
+ .minValue(effectiveType == null ? null : effectiveType.toComparable(_min))
+ .maxValue(effectiveType == null ? null : effectiveType.toComparable(_max))
+ .ndv(_maxNdv, StatConfidence.ESTIMATED)
+ .minTrusted(!_anyUntrustedMin && effectiveType != null)
+ .avgBytesPerValue(_avgBytesDocs > 0 ? _weightedAvgBytes / _avgBytesDocs : -1)
+ .nullFraction(_nullFractionDocs > 0 ? _weightedNullFraction / _nullFractionDocs : -1)
+ .build();
+ }
+ }
+
+ /// Returns the smaller of two stored values under `type`; `null` means unknown, so the other
+ /// value wins.
+ @Nullable
+ public static String minOf(@Nullable String a, @Nullable String b, ColumnValueType type) {
+ if (a == null) {
+ return b;
+ }
+ if (b == null) {
+ return a;
+ }
+ return type.compare(a, b) <= 0 ? a : b;
+ }
+
+ /// Returns the larger of two stored values under `type`; `null` means unknown, so the other
+ /// value wins.
+ @Nullable
+ public static String maxOf(@Nullable String a, @Nullable String b, ColumnValueType type) {
+ if (a == null) {
+ return b;
+ }
+ if (b == null) {
+ return a;
+ }
+ return type.compare(a, b) >= 0 ? a : b;
+ }
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsStore.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsStore.java
new file mode 100644
index 00000000000..0326e98f2cd
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsStore.java
@@ -0,0 +1,153 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+import java.io.Closeable;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.OptionalLong;
+import java.util.Set;
+import javax.annotation.Nullable;
+
+
+/// Broker-local persistence for table and segment statistics used by the cost-based query planner.
+///
+/// Durability is implementation-defined: a durable store survives broker restarts warm, while a
+/// non-durable store restarts empty and relies on the caller's reconciliation path
+/// ([#getSegmentCrcs]) to re-collect, trading restart re-collection cost for zero disk usage.
+///
+/// Threading: writes arrive from a pool that processes segment-assignment changes table by table,
+/// so calls are serialized per `tableNameWithType` but NOT across tables — implementations must be
+/// safe for concurrent writers to different tables, and for readers at any time. READ failures must
+/// be cheap to detect so callers can degrade to a no-stats path; callers must never fail a query
+/// because of a store error.
+///
+/// Lifecycle: call [#init()] once before any other method; call [#close()] to release resources.
+///
+/// Thread-safety: see the threading note above — concurrent readers plus concurrent writers to
+/// distinct tables.
+public interface StatsStore extends Closeable {
+
+ /// Opens the store and migrates the schema if necessary.
+ ///
+ /// On unrecoverable corruption, implementations must drop and recreate an empty store rather
+ /// than propagating an error.
+ ///
+ /// @throws StatsStoreException if the store cannot be opened or initialized
+ void init()
+ throws StatsStoreException;
+
+ /// Inserts or updates segment-level statistics for the given table.
+ ///
+ /// @param tableNameWithType fully-qualified table name including type suffix
+ /// @param rows segment statistics rows to upsert
+ /// @throws StatsStoreException if the write fails
+ void upsertSegmentStats(String tableNameWithType, List<SegmentStatsRow> rows)
+ throws StatsStoreException;
+
+ /// Inserts or updates per-column statistics for individual segments of the given table.
+ ///
+ /// @param tableNameWithType fully-qualified table name including type suffix
+ /// @param rows segment column statistics rows to upsert
+ /// @throws StatsStoreException if the write fails
+ void upsertSegmentColumnStats(String tableNameWithType, List<SegmentColumnStatsRow> rows)
+ throws StatsStoreException;
+
+ /// Removes all stored statistics for the specified segments of the given table.
+ ///
+ /// @param tableNameWithType fully-qualified table name including type suffix
+ /// @param segmentNames names of segments to remove
+ /// @throws StatsStoreException if the removal fails
+ void removeSegments(String tableNameWithType, Collection<String> segmentNames)
+ throws StatsStoreException;
+
+ /// Returns a map from segment name to CRC for all segments of the given table.
+ ///
+ /// Used for restart reconciliation to detect stale or missing entries.
+ ///
+ /// @param tableNameWithType fully-qualified table name including type suffix
+ /// @return map of segment name to CRC; empty map if no segments are stored
+ /// @throws StatsStoreException if the read fails
+ Map<String, Long> getSegmentCrcs(String tableNameWithType)
+ throws StatsStoreException;
+
+ /// Returns aggregated table-level statistics derived from all non-consuming segments, or
+ /// `null` if no statistics are available.
+ ///
+ /// @param tableNameWithType fully-qualified table name including type suffix
+ /// @throws StatsStoreException if the read fails
+ @Nullable
+ TableStatistics getTableStats(String tableNameWithType)
+ throws StatsStoreException;
+
+ /// Returns per-column statistics aggregated across all segments for the given table and column,
+ /// or `null` if no statistics are available.
+ ///
+ /// @param tableNameWithType fully-qualified table name including type suffix
+ /// @param columnName name of the column
+ /// @throws StatsStoreException if the read fails
+ @Nullable
+ ColumnStatistics getColumnStats(String tableNameWithType, String columnName)
+ throws StatsStoreException;
+
+ /// Returns an estimate of the number of rows whose time column falls in the half-open interval
+ /// `[startMs, endMs)`, or an empty optional if the estimate cannot be produced.
+ ///
+ /// @param tableNameWithType fully-qualified table name including type suffix
+ /// @param startMs start of the time range, inclusive, in epoch milliseconds
+ /// @param endMs end of the time range, exclusive, in epoch milliseconds
+ /// @throws StatsStoreException if the read fails
+ OptionalLong estimateRowsInTimeRange(String tableNameWithType, long startMs, long endMs)
+ throws StatsStoreException;
+
+ /// Returns every table this store currently holds statistics for.
+ ///
+ /// Exists so a caller can find tables it no longer serves: per-table cleanup is otherwise driven
+ /// by an event (a routing entry being removed), which a broker cannot observe for a table dropped
+ /// while it was down. Without this the rows of such a table would stay in a durable store forever.
+ ///
+ /// @return fully-qualified table names, including type suffix; empty if the store holds nothing
+ /// @throws StatsStoreException if the read fails
+ Set<String> getTables()
+ throws StatsStoreException;
+
+ /// Removes all stored statistics for the given table.
+ ///
+ /// @param tableNameWithType fully-qualified table name including type suffix
+ /// @throws StatsStoreException if the purge fails
+ void purgeTable(String tableNameWithType)
+ throws StatsStoreException;
+
+ /// Removes all stored statistics for all tables.
+ ///
+ /// @throws StatsStoreException if the purge fails
+ void purgeAll()
+ throws StatsStoreException;
+
+ /// Returns `true` if the given table has at least one consuming (REALTIME IN_PROGRESS)
+ /// segment in the store.
+ ///
+ /// Used to detect whether realtime row counts may undercount fresh, un-committed data.
+ ///
+ /// @param tableNameWithType fully-qualified table name including type suffix
+ /// @throws StatsStoreException if the read fails
+ boolean hasConsumingSegments(String tableNameWithType)
+ throws StatsStoreException;
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsStoreException.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsStoreException.java
new file mode 100644
index 00000000000..7c9b2b6fb77
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsStoreException.java
@@ -0,0 +1,44 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+
+/// Checked exception thrown by [StatsStore] operations when a storage error occurs.
+///
+/// Callers of [StatsStore] should catch this exception and degrade gracefully to a
+/// no-stats path rather than failing the query.
+///
+/// Thread-safety: exception objects are not shared; no concurrency requirements.
+public class StatsStoreException extends Exception {
+
+ /// Constructs a new [StatsStoreException] with the given message.
+ ///
+ /// @param message description of the error
+ public StatsStoreException(String message) {
+ super(message);
+ }
+
+ /// Constructs a new [StatsStoreException] with the given message and cause.
+ ///
+ /// @param message description of the error
+ /// @param cause the underlying cause
+ public StatsStoreException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsStoreProvider.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsStoreProvider.java
new file mode 100644
index 00000000000..12802288640
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/StatsStoreProvider.java
@@ -0,0 +1,45 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+import java.util.Map;
+
+
+/// Creates a [StatsStore] for a configured name, discovered through [java.util.ServiceLoader].
+///
+/// Implementations declare their own name, so operators configure a stable identifier rather than a
+/// class name: renaming or moving the implementation must not invalidate an operator's
+/// configuration. Register one by listing it in
+/// `META-INF/services/org.apache.pinot.query.planner.spi.stats.StatsStoreProvider`.
+///
+/// Thread-safety: providers are discovered once and shared, so implementations must be thread-safe.
+public interface StatsStoreProvider {
+
+ /// Returns the identifier operators use to select this store. Must be unique across all providers
+ /// on the classpath and stable across releases, since it appears in configuration.
+ String getName();
+
+ /// Creates a store; the caller initializes it via [StatsStore#init()].
+ ///
+ /// @param properties statistics-related broker configuration, with keys stripped of the
+ /// `pinot.broker.stats.` prefix (for example `dir`)
+ /// @throws StatsStoreException if the store cannot be created from this configuration
+ StatsStore create(Map<String, String> properties)
+ throws StatsStoreException;
+}
diff --git a/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/TableStatistics.java b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/TableStatistics.java
new file mode 100644
index 00000000000..f9a35fe7374
--- /dev/null
+++ b/pinot-query-planner-spi/src/main/java/org/apache/pinot/query/planner/spi/stats/TableStatistics.java
@@ -0,0 +1,116 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+
+/// Immutable aggregate statistics for a table, used by the cost-based query planner.
+///
+/// Instances are created via [#builder()]. Unknown numeric fields are represented
+/// by `-1`; unknown timestamp fields by `0`.
+///
+/// Thread-safety: immutable; safe for concurrent access.
+public class TableStatistics {
+ private final long _rowCount;
+ private final StatConfidence _rowCountConfidence;
+ private final long _tableSizeBytes;
+ private final StatConfidence _sizeConfidence;
+ private final long _updatedAtMs;
+
+ private TableStatistics(Builder builder) {
+ _rowCount = builder._rowCount;
+ _rowCountConfidence = builder._rowCountConfidence;
+ _tableSizeBytes = builder._tableSizeBytes;
+ _sizeConfidence = builder._sizeConfidence;
+ _updatedAtMs = builder._updatedAtMs;
+ }
+
+ /// Returns a new [Builder] for constructing [TableStatistics] instances.
+ public static Builder builder() {
+ return new Builder();
+ }
+
+ /// Returns the estimated number of rows in the table, or `-1` if unknown.
+ public long getRowCount() {
+ return _rowCount;
+ }
+
+ /// Returns the confidence level of the [#getRowCount()] value.
+ public StatConfidence getRowCountConfidence() {
+ return _rowCountConfidence;
+ }
+
+ /// Returns the estimated total size of the table in bytes, or `-1` if unknown.
+ public long getTableSizeBytes() {
+ return _tableSizeBytes;
+ }
+
+ /// Returns the confidence level of the [#getTableSizeBytes()] value.
+ public StatConfidence getSizeConfidence() {
+ return _sizeConfidence;
+ }
+
+ /// Returns the epoch-millisecond timestamp at which these statistics were last updated,
+ /// or `0` if unknown.
+ public long getUpdatedAtMs() {
+ return _updatedAtMs;
+ }
+
+ /// Builder for [TableStatistics].
+ ///
+ /// Default values: numeric fields default to `-1` (unknown),
+ /// confidence fields default to [StatConfidence#UNKNOWN],
+ /// timestamp fields default to `0` (unknown).
+ ///
+ /// Thread-safety: not thread-safe; use from a single thread.
+ public static class Builder {
+ private long _rowCount = -1;
+ private StatConfidence _rowCountConfidence = StatConfidence.UNKNOWN;
+ private long _tableSizeBytes = -1;
+ private StatConfidence _sizeConfidence = StatConfidence.UNKNOWN;
+ private long _updatedAtMs = 0;
+
+ private Builder() {
+ }
+
+ /// Sets the row count and its confidence level.
+ public Builder rowCount(long rowCount, StatConfidence confidence) {
+ _rowCount = rowCount;
+ _rowCountConfidence = confidence;
+ return this;
+ }
+
+ /// Sets the table size in bytes and its confidence level.
+ public Builder tableSizeBytes(long tableSizeBytes, StatConfidence confidence) {
+ _tableSizeBytes = tableSizeBytes;
+ _sizeConfidence = confidence;
+ return this;
+ }
+
+ /// Sets the epoch-millisecond timestamp of the last statistics update.
+ public Builder updatedAtMs(long updatedAtMs) {
+ _updatedAtMs = updatedAtMs;
+ return this;
+ }
+
+ /// Builds the immutable [TableStatistics] instance.
+ public TableStatistics build() {
+ return new TableStatistics(this);
+ }
+ }
+}
diff --git a/pinot-query-planner-spi/src/test/java/org/apache/pinot/query/planner/spi/stats/ColumnValueTypeTest.java b/pinot-query-planner-spi/src/test/java/org/apache/pinot/query/planner/spi/stats/ColumnValueTypeTest.java
new file mode 100644
index 00000000000..bf4361b1dfb
--- /dev/null
+++ b/pinot-query-planner-spi/src/test/java/org/apache/pinot/query/planner/spi/stats/ColumnValueTypeTest.java
@@ -0,0 +1,106 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+import java.math.BigDecimal;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Covers the ordering dispatch and its degradation paths. Both matter on the query-planning path:
+/// a throw here would fail a query rather than cost it an estimate.
+public class ColumnValueTypeTest {
+
+ @DataProvider(name = "orderedPairs")
+ public Object[][] orderedPairs() {
+ // Each row is a value that must order BELOW the next, under that type's own ordering.
+ return new Object[][]{
+ {ColumnValueType.LONG, "9", "10"},
+ // Beyond 2^53, where a double round-trip would lose the distinction entirely.
+ {ColumnValueType.LONG, "9007199254740993", "9007199254740994"},
+ {ColumnValueType.DOUBLE, "1.5", "10.5"},
+ {ColumnValueType.DOUBLE, "-2.5", "-1.5"},
+ {ColumnValueType.BIG_DECIMAL, "9.10", "10.01"},
+ // Lexical, which is exactly where a numeric ordering would disagree.
+ {ColumnValueType.STRING, "10", "9"},
+ };
+ }
+
+ @Test(dataProvider = "orderedPairs")
+ public void testCompareOrdersByTheDeclaredType(ColumnValueType type, String lower, String higher) {
+ assertTrue(type.compare(lower, higher) < 0, type + ": " + lower + " should order below " + higher);
+ assertTrue(type.compare(higher, lower) > 0, type + ": " + higher + " should order above " + lower);
+ assertEquals(type.compare(lower, lower), 0);
+ }
+
+ @DataProvider(name = "types")
+ public Object[][] types() {
+ return new Object[][]{
+ {ColumnValueType.LONG}, {ColumnValueType.DOUBLE}, {ColumnValueType.BIG_DECIMAL}, {ColumnValueType.STRING}
+ };
+ }
+
+ @Test(dataProvider = "types")
+ public void testCompareFallsBackToLexicalOnMalformedValues(ColumnValueType type) {
+ // A stored value that does not parse must degrade, not throw.
+ assertEquals(type.compare("abc", "abd"), "abc".compareTo("abd"));
+ assertEquals(type.compare("abc", "abc"), 0);
+ // One side malformed is still enough to force the lexical path.
+ assertEquals(type.compare("1", "abc"), "1".compareTo("abc"));
+ }
+
+ @Test
+ public void testToComparableDeserializesByType() {
+ assertEquals(ColumnValueType.LONG.toComparable("42"), 42L);
+ assertEquals(ColumnValueType.DOUBLE.toComparable("42.5"), 42.5d);
+ assertEquals(ColumnValueType.BIG_DECIMAL.toComparable("42.50"), new BigDecimal("42.50"));
+ assertEquals(ColumnValueType.STRING.toComparable("42"), "42");
+ }
+
+ @Test(dataProvider = "types")
+ public void testToComparableReturnsRawValueWhenUnparseable(ColumnValueType type) {
+ assertEquals(type.toComparable("not-a-number"), "not-a-number");
+ }
+
+ @Test(dataProvider = "types")
+ public void testToComparableKeepsNull(ColumnValueType type) {
+ assertNull(type.toComparable(null));
+ }
+
+ @Test
+ public void testFromNameResolvesEveryConstant() {
+ for (ColumnValueType type : ColumnValueType.values()) {
+ assertEquals(ColumnValueType.fromName(type.name()), type);
+ }
+ }
+
+ @Test
+ public void testFromNameYieldsNullForUnknownNames() {
+ // A store written by a newer build may carry a name this one does not have. That must cost
+ // precision (null means "ordering unknown"), not throw out of the planning path.
+ assertNull(ColumnValueType.fromName("TIMESTAMP_WITH_TIMEZONE"));
+ assertNull(ColumnValueType.fromName(""));
+ assertNull(ColumnValueType.fromName("long"));
+ assertNull(ColumnValueType.fromName(null));
+ }
+}
diff --git a/pinot-query-planner-spi/src/test/java/org/apache/pinot/query/planner/spi/stats/StatsAggregationsTest.java b/pinot-query-planner-spi/src/test/java/org/apache/pinot/query/planner/spi/stats/StatsAggregationsTest.java
new file mode 100644
index 00000000000..128df01a89a
--- /dev/null
+++ b/pinot-query-planner-spi/src/test/java/org/apache/pinot/query/planner/spi/stats/StatsAggregationsTest.java
@@ -0,0 +1,148 @@
+/**
+ * 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.pinot.query.planner.spi.stats;
+
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Covers the rollup semantics both stores fold through, so they cannot drift apart.
+public class StatsAggregationsTest {
+
+ private static final String COLUMN = "col";
+
+ private static SegmentColumnStatsRow row(String segment, long ndv, String min, String max, boolean minTrusted,
+ double avgBytes, double nullFraction, ColumnValueType valueType) {
+ return new SegmentColumnStatsRow(segment, COLUMN, ndv, min, max, minTrusted, avgBytes, nullFraction, valueType);
+ }
+
+ @Test
+ public void testEmptyAccumulatorReportsNoStatistics() {
+ assertTrue(new StatsAggregations.ColumnStatsAccumulator().isEmpty());
+ }
+
+ @Test
+ public void testBoundsUseTheRecordedOrdering() {
+ StatsAggregations.ColumnStatsAccumulator acc = new StatsAggregations.ColumnStatsAccumulator();
+ acc.add(100, row("s1", 5, "9", "9", true, 4, 0.0, ColumnValueType.LONG));
+ acc.add(100, row("s2", 7, "10", "10", true, 4, 0.0, ColumnValueType.LONG));
+ ColumnStatistics stats = acc.build(COLUMN);
+ // Lexically "10" < "9"; numerically it is not. The recorded type decides.
+ assertEquals(stats.getMinValue(), 9L);
+ assertEquals(stats.getMaxValue(), 10L);
+ assertTrue(stats.isMinTrusted());
+ }
+
+ @Test
+ public void testStringColumnKeepsLexicalOrdering() {
+ StatsAggregations.ColumnStatsAccumulator acc = new StatsAggregations.ColumnStatsAccumulator();
+ acc.add(100, row("s1", 5, "9", "9", true, 4, 0.0, ColumnValueType.STRING));
+ acc.add(100, row("s2", 7, "10", "10", true, 4, 0.0, ColumnValueType.STRING));
+ ColumnStatistics stats = acc.build(COLUMN);
+ assertEquals(stats.getMinValue(), "10");
+ assertEquals(stats.getMaxValue(), "9");
+ }
+
+ @Test
+ public void testUnrecordedTypeDropsBothBounds() {
+ StatsAggregations.ColumnStatsAccumulator acc = new StatsAggregations.ColumnStatsAccumulator();
+ acc.add(100, row("s1", 5, "1", "500", true, 4, 0.0, null));
+ ColumnStatistics stats = acc.build(COLUMN);
+ // Bounds folded without a known ordering are neither a true minimum nor a true maximum, and
+ // isMinTrusted says nothing about the maximum -- so neither is reported at all.
+ assertNull(stats.getMinValue());
+ assertNull(stats.getMaxValue());
+ assertFalse(stats.isMinTrusted());
+ // NDV is unaffected: it needs no ordering.
+ assertEquals(stats.getNdv(), 5);
+ }
+
+ @Test
+ public void testConflictingTypesAcrossSegmentsDropBothBounds() {
+ StatsAggregations.ColumnStatsAccumulator acc = new StatsAggregations.ColumnStatsAccumulator();
+ acc.add(100, row("s1", 5, "1", "500", true, 4, 0.0, ColumnValueType.LONG));
+ acc.add(100, row("s2", 5, "a", "z", true, 4, 0.0, ColumnValueType.STRING));
+ ColumnStatistics stats = acc.build(COLUMN);
+ assertNull(stats.getMinValue());
+ assertNull(stats.getMaxValue());
+ assertFalse(stats.isMinTrusted());
+ }
+
+ @Test
+ public void testUntrustedMinInAnySegmentTaintsTheAggregate() {
+ StatsAggregations.ColumnStatsAccumulator acc = new StatsAggregations.ColumnStatsAccumulator();
+ acc.add(100, row("s1", 5, "1", "5", true, 4, 0.0, ColumnValueType.LONG));
+ acc.add(100, row("s2", 5, "2", "6", false, 4, 0.0, ColumnValueType.LONG));
+ assertFalse(acc.build(COLUMN).isMinTrusted());
+ }
+
+ @Test
+ public void testWeightedAveragesAreDocumentWeighted() {
+ StatsAggregations.ColumnStatsAccumulator acc = new StatsAggregations.ColumnStatsAccumulator();
+ acc.add(300, row("s1", 5, "1", "5", true, 8, 0.4, ColumnValueType.LONG));
+ acc.add(100, row("s2", 5, "1", "5", true, 4, 0.0, ColumnValueType.LONG));
+ ColumnStatistics stats = acc.build(COLUMN);
+ assertEquals(stats.getAvgBytesPerValue(), (8 * 300 + 4 * 100) / 400.0, 1e-9);
+ assertEquals(stats.getNullFraction(), (0.4 * 300) / 400.0, 1e-9);
+ }
+
+ @Test
+ public void testUnknownSentinelsAreExcludedFromAverages() {
+ StatsAggregations.ColumnStatsAccumulator acc = new StatsAggregations.ColumnStatsAccumulator();
+ acc.add(1000, row("s1", 5, "1", "5", true, -1, -1, ColumnValueType.LONG));
+ acc.add(1000, row("s2", 5, "1", "5", true, 4, 0.2, ColumnValueType.LONG));
+ ColumnStatistics stats = acc.build(COLUMN);
+ // Weighting the -1 sentinel in as a measurement would yield 1.5 bytes and a NEGATIVE null
+ // fraction -- neither a legal value nor the sentinel a consumer tests for.
+ assertEquals(stats.getAvgBytesPerValue(), 4.0, 1e-9);
+ assertEquals(stats.getNullFraction(), 0.2, 1e-9);
+ }
+
+ @Test
+ public void testAveragesReportUnknownWhenEverySegmentIsUnknown() {
+ StatsAggregations.ColumnStatsAccumulator acc = new StatsAggregations.ColumnStatsAccumulator();
+ acc.add(1000, row("s1", 5, "1", "5", true, -1, -1, ColumnValueType.LONG));
+ ColumnStatistics stats = acc.build(COLUMN);
+ assertEquals(stats.getAvgBytesPerValue(), -1.0, 1e-9);
+ assertEquals(stats.getNullFraction(), -1.0, 1e-9);
+ }
+
+ @Test
+ public void testNdvTakesTheLargestSegmentValue() {
+ StatsAggregations.ColumnStatsAccumulator acc = new StatsAggregations.ColumnStatsAccumulator();
+ acc.add(100, row("s1", -1, "1", "5", true, 4, 0.0, ColumnValueType.LONG));
+ acc.add(100, row("s2", 12, "1", "5", true, 4, 0.0, ColumnValueType.LONG));
+ // MAX over segments is a lower bound on the table-wide NDV; the -1 sentinel cannot win it.
+ assertEquals(acc.build(COLUMN).getNdv(), 12);
+ }
+
+ @Test
+ public void testOverlapRowsInterpolatesPartialOverlap() {
+ // Fully contained.
+ assertEquals(StatsAggregations.overlapRows(100, 10, 20, 0, 100), 100);
+ // No overlap at all.
+ assertEquals(StatsAggregations.overlapRows(100, 10, 20, 30, 40), 0);
+ // Half the segment's span falls in the range, so half its rows are attributed to it.
+ assertEquals(StatsAggregations.overlapRows(100, 0, 100, 0, 50), 50);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org