gortiz commented on code in PR #18741: URL: https://github.com/apache/pinot/pull/18741#discussion_r3665866314
########## pinot-broker/src/main/java/org/apache/pinot/broker/stats/SqliteStatsStore.java: ########## @@ -0,0 +1,715 @@ +/** + * 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.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.OptionalLong; +import javax.annotation.Nullable; +import org.apache.pinot.query.planner.spi.stats.ColumnStatistics; +import org.apache.pinot.query.planner.spi.stats.StatConfidence; +import org.apache.pinot.query.planner.spi.stats.TableStatistics; +import org.flywaydb.core.Flyway; +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 ([#READ_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 and migrate the database. 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); + + /// Number of read connections in the shared pool. + static final int READ_POOL_SIZE = 4; + + private static final String DB_FILE_NAME = "broker-stats.sqlite"; + private static final String MIGRATION_LOCATION = "classpath:db/broker-stats-migration"; + + // 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,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,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=?"; + + private static final String SQL_TABLE_STATS = + "SELECT SUM(total_docs),SUM(size_bytes),MAX(updated_at_ms),COUNT(*) " + + "FROM segment_stats WHERE table_name=? AND consuming=0"; + + 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 " + + "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"; + + 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"; + + private static final String SQL_HAS_CONSUMING = + "SELECT 1 FROM segment_stats WHERE table_name=? AND consuming=1 LIMIT 1"; + + 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; + + /// Write connection — all mutations go through this; guarded by _writeLock. + private Connection _writeConn; + 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 [#READ_POOL_SIZE]. + private final java.util.concurrent.ArrayBlockingQueue<Connection> _readPool = + new java.util.concurrent.ArrayBlockingQueue<>(READ_POOL_SIZE); + + 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) { + _dbDirectory = dbDirectory; + _dbPath = dbDirectory.resolve(DB_FILE_NAME); + } + + // --------------------------------------------------------------------------- + // Lifecycle + // --------------------------------------------------------------------------- + + @Override + public void init() + throws StatsStoreException { + try { + openAndMigrate(); + } catch (Exception firstEx) { + LOGGER.warn( + "Failed to open stats store at {}; deleting and retrying from scratch. Cause: {}", + _dbPath, firstEx.getMessage(), firstEx); + deleteDbFiles(); + try { + openAndMigrate(); + } catch (Exception secondEx) { + throw new StatsStoreException( + "Cannot initialise SqliteStatsStore at " + _dbPath, secondEx); + } + } + } + + private void openAndMigrate() + throws Exception { + Files.createDirectories(_dbDirectory); + String jdbcUrl = "jdbc:sqlite:" + _dbPath.toAbsolutePath(); + + // Run Flyway migrations first (uses its own connection internally) + Flyway flyway = Flyway.configure() + .dataSource(jdbcUrl, null, null) + .locations(MIGRATION_LOCATION) + .load(); + flyway.migrate(); + + // 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"); + } + conn.setAutoCommit(false); + _writeConn = conn; + + // Open read-only connections for the pool + for (int i = 0; i < READ_POOL_SIZE; 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"); + } + _readPool.offer(rConn); + } + } + + /// 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; + } + Connection c; + while ((c = _readPool.poll()) != null) { + closeQuietly(c); + } + } + + 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 = _writeConn.prepareStatement(SQL_UPSERT_SEGMENT)) { + for (SegmentStatsRow row : rows) { + ps.setString(1, tableNameWithType); + ps.setString(2, row.getSegmentName()); + ps.setLong(3, row.getCrc()); + ps.setLong(4, row.getTotalDocs()); + ps.setLong(5, row.getSizeBytes()); + ps.setLong(6, row.getStartTimeMs()); + ps.setLong(7, row.getEndTimeMs()); + ps.setInt(8, row.isConsuming() ? 1 : 0); + ps.setLong(9, now); + ps.addBatch(); + } + ps.executeBatch(); + } + _writeConn.commit(); + } catch (SQLException e) { + rollbackQuietly(_writeConn); + 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 = _writeConn.prepareStatement(SQL_UPSERT_COL)) { + for (SegmentColumnStatsRow row : rows) { + ps.setString(1, tableNameWithType); + ps.setString(2, row.getSegmentName()); + ps.setString(3, row.getColumnName()); + ps.setLong(4, row.getNdv()); + ps.setString(5, row.getMinValue()); + ps.setString(6, row.getMaxValue()); + ps.setInt(7, row.isMinTrusted() ? 1 : 0); + ps.setDouble(8, row.getAvgBytesPerValue()); + ps.setDouble(9, row.getNullFraction()); + ps.setLong(10, now); + ps.addBatch(); + } + ps.executeBatch(); + } + _writeConn.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 = _writeConn.prepareStatement(SQL_DELETE_SEGMENT); + PreparedStatement psCol = _writeConn.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(); + } + _writeConn.commit(); + } catch (SQLException e) { + rollbackQuietly(_writeConn); + throw new StatsStoreException("removeSegments failed for " + tableNameWithType, e); + } + } + } + + @Override + public boolean hasConsumingSegments(String tableNameWithType) + throws StatsStoreException { + checkOpen(); + Connection conn = borrowReadConn(); + try { + try (PreparedStatement ps = conn.prepareStatement(SQL_HAS_CONSUMING)) { + ps.setString(1, tableNameWithType); + try (ResultSet rs = ps.executeQuery()) { + return rs.next(); + } + } + } catch (SQLException e) { + throw new StatsStoreException("hasConsumingSegments failed for " + tableNameWithType, e); + } finally { + returnReadConn(conn); + } + } + + @Override + public void purgeTable(String tableNameWithType) + throws StatsStoreException { + checkOpen(); + synchronized (_writeLock) { + try { + try (PreparedStatement psSeg = _writeConn.prepareStatement(SQL_PURGE_TABLE_SEG); + PreparedStatement psCol = _writeConn.prepareStatement(SQL_PURGE_TABLE_COL)) { + psSeg.setString(1, tableNameWithType); + psSeg.executeUpdate(); + psCol.setString(1, tableNameWithType); + psCol.executeUpdate(); + } + _writeConn.commit(); + } catch (SQLException e) { + rollbackQuietly(_writeConn); + throw new StatsStoreException("purgeTable failed for " + tableNameWithType, e); + } + } + } + + @Override + public void purgeAll() + throws StatsStoreException { + checkOpen(); + synchronized (_writeLock) { + try { + try (Statement st = _writeConn.createStatement()) { + st.execute(SQL_PURGE_ALL_SEG); + st.execute(SQL_PURGE_ALL_COL); + } + _writeConn.commit(); + } catch (SQLException e) { + rollbackQuietly(_writeConn); + throw new StatsStoreException("purgeAll failed", e); + } + } + } + + // --------------------------------------------------------------------------- + // Read operations + // --------------------------------------------------------------------------- + + @Override + public Map<String, Long> getSegmentCrcs(String tableNameWithType) + throws StatsStoreException { + checkOpen(); + Connection conn = borrowReadConn(); + try { + Map<String, Long> result = new HashMap<>(); + try (PreparedStatement ps = conn.prepareStatement(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 { + returnReadConn(conn); + } + } + + @Override + @Nullable + public TableStatistics getTableStats(String tableNameWithType) + throws StatsStoreException { + checkOpen(); + Connection conn = borrowReadConn(); + try { + try (PreparedStatement ps = conn.prepareStatement(SQL_TABLE_STATS)) { + ps.setString(1, tableNameWithType); + try (ResultSet rs = ps.executeQuery()) { + if (!rs.next()) { + return null; + } + long totalDocs = rs.getLong(1); + long sizeBytes = rs.getLong(2); + long maxUpdatedAt = rs.getLong(3); + long count = rs.getLong(4); + if (count == 0) { + return null; + } + return TableStatistics.builder() + .rowCount(totalDocs, StatConfidence.EXACT) + .tableSizeBytes(sizeBytes, StatConfidence.EXACT) + .updatedAtMs(maxUpdatedAt) + .build(); + } + } + } catch (SQLException e) { + throw new StatsStoreException("getTableStats failed for " + tableNameWithType, e); + } finally { + returnReadConn(conn); + } + } + + /// 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(); + Connection conn = borrowReadConn(); + try { + List<long[]> docsNdvTrusted = new ArrayList<>(); + List<String[]> minMax = new ArrayList<>(); + List<double[]> avgBytesNullFrac = new ArrayList<>(); + + try (PreparedStatement ps = conn.prepareStatement(SQL_COL_STATS)) { + ps.setString(1, tableNameWithType); + ps.setString(2, columnName); + try (ResultSet rs = ps.executeQuery()) { + while (rs.next()) { + long docs = rs.getLong(1); + long ndv = rs.getLong(2); + String minVal = rs.getString(3); + String maxVal = rs.getString(4); + boolean minTrusted = rs.getInt(5) != 0; + double avgBytes = rs.getDouble(6); + double nullFrac = rs.getDouble(7); + + docsNdvTrusted.add(new long[]{docs, ndv, minTrusted ? 1L : 0L}); + minMax.add(new String[]{minVal, maxVal}); + avgBytesNullFrac.add(new double[]{avgBytes, nullFrac}); + } + } + } + + if (docsNdvTrusted.isEmpty()) { + return null; + } + + // Aggregate + long maxNdv = -1; + boolean anyUntrustedMin = false; + long totalDocs = 0; + double weightedAvgBytes = 0; + double weightedNullFrac = 0; + String globalMin = null; + String globalMax = null; + + for (int i = 0; i < docsNdvTrusted.size(); i++) { + long docs = docsNdvTrusted.get(i)[0]; + long ndv = docsNdvTrusted.get(i)[1]; + boolean trusted = docsNdvTrusted.get(i)[2] != 0; + double avgBytes = avgBytesNullFrac.get(i)[0]; + double nullFrac = avgBytesNullFrac.get(i)[1]; + String minVal = minMax.get(i)[0]; + String maxVal = minMax.get(i)[1]; + + if (ndv > maxNdv) { + maxNdv = ndv; + } + if (!trusted) { + anyUntrustedMin = true; + } + totalDocs += docs; + weightedAvgBytes += avgBytes * docs; + weightedNullFrac += nullFrac * docs; + globalMin = minOf(globalMin, minVal); + globalMax = maxOf(globalMax, maxVal); + } + + double finalAvgBytes = totalDocs > 0 ? weightedAvgBytes / totalDocs : -1; + double finalNullFrac = totalDocs > 0 ? weightedNullFrac / totalDocs : -1; + + // Build comparable min/max values + Comparable<?> minComparable = toComparable(globalMin); + Comparable<?> maxComparable = toComparable(globalMax); + + return ColumnStatistics.builder() + .columnName(columnName) + .ndv(maxNdv, StatConfidence.ESTIMATED) + .minValue(minComparable) + .maxValue(maxComparable) + .minTrusted(!anyUntrustedMin) + .avgBytesPerValue(finalAvgBytes) + .nullFraction(finalNullFrac) + .build(); + } catch (SQLException e) { + throw new StatsStoreException( + "getColumnStats failed for " + tableNameWithType + "." + columnName, e); + } finally { + returnReadConn(conn); + } + } + + @Override + public OptionalLong estimateRowsInTimeRange(String tableNameWithType, long startMs, long endMs) + throws StatsStoreException { + checkOpen(); + Connection conn = borrowReadConn(); + try { + long totalRows = 0; + boolean hasAnyRow = false; + + try (PreparedStatement ps = conn.prepareStatement(SQL_TIME_RANGE)) { + ps.setString(1, 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; + + // Unknown times (-1) → conservative: include all docs + if (segStart == -1 || segEnd == -1) { + totalRows += docs; + continue; + } + + // Check overlap with [startMs, endMs) + if (segEnd <= startMs || segStart >= endMs) { + // No overlap + continue; Review Comment: The main reason the estimation lives in Java is architectural rather than stylistic: `estimateRowsInTimeRange` is a primitive that `LogicalTableStatsResolver` invokes with caller-supplied bounds — most importantly the **hybrid-table merge**, which calls it once per physical table with ranges derived from the **live time boundary** (`offline [MIN, boundary)` + `realtime [boundary, MAX)`). The boundary comes from a broker-side provider (it moves as segments commit), and the rest of the pipeline — upsert/dedup and consuming-segment confidence adjustments from table configs, per-side fallbacks, cross-table confidence merging — consumes inputs that aren't in SQLite at all. Pushing the interpolation into SQL wouldn't remove that layer; it would split one estimator's semantics across two languages with the combination still in Java. So the store stays a dumb per-segment source of truth (which also keeps the `-1`-sentinel / zero-length / rounding edge cases directly unit-testable). On the efficiency itself: the query already narrows to `table_name=? AND consuming=0` and three columns, so what crosses JDBC is one `(docs, start, end)` triple per committed segment of a single table — and SQLite scans the same rows whether the arithmetic runs in SQL or Java. That said, you're right that the *overlap predicate* belongs in the WHERE clause — done in 5d23fd6986. One subtlety made it slightly more than a one-liner: pruning in SQL breaks the `of(0)` ("no overlapping segments") vs `empty()` ("no stats") distinction, and doing an existence check as a *second* statement would race with concurrent stats writes. So the predicate and an existence sentinel live in the **same statement** (`UNION ALL … WHERE EXISTS`), keeping the whole answer on one snapshot. Also added tests for the consuming-only and boundary-adjacency cases. ########## pinot-query-planner/src/main/java/org/apache/pinot/calcite/plan/PinotRelOptCost.java: ########## @@ -0,0 +1,316 @@ +/** + * 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.calcite.plan; + +import java.util.Objects; +import org.apache.calcite.plan.RelOptCost; +import org.apache.calcite.plan.RelOptCostFactory; + + +/// Pinot-specific implementation of [RelOptCost] for the multi-stage query engine (MSE). +/// +/// ### Ordering semantics (rows-dominated lexicographic) +/// Cost comparison uses a strict (rows, cpu, io) lexicographic order: +/// 1. Row count is the primary key — plans that process fewer rows are cheaper. +/// 1. CPU cost breaks ties when row counts are equal. +/// 1. IO cost breaks ties when both rows and CPU are equal. +/// +/// ### Deliberate difference from Calcite's `VolcanoCost` +/// Calcite's `VolcanoCost.isLe` compares *only* row count: two costs with the same +/// number of rows are considered equal regardless of their cpu or io values. This means +/// `VolcanoCost` has no way to break ties when row counts match, which can leave join-order +/// choices arbitrary. Review Comment: We can't use it directly: `VolcanoCost` is package-private in `org.apache.calcite.plan.volcano` — it's only reachable as `VolcanoPlanner`'s internal default, and the reorder phase runs as a dedicated Hep sub-phase. The public off-the-shelf class is `RelOptCostImpl`, which is strictly weaker: rows-only, with `getCpu()`/`getIo()` hardwired to 0. And even where VolcanoCost is usable, its `isLe`/`equals` compare only `rowCount`, so alternatives with equal row counts tie and the winner falls out of iteration order. The lexicographic rows→cpu→io comparison here isn't a performance claim that needs proving in a follow-up — it's **determinism**: same query, same stats, same plan on every run and JVM. The plan-level tests in this PR depend on that stability, and "the plan doesn't depend on accidents of ordering" is the same property that motivates the whole reorder phase (no write-order sensitivity). It's also the one cost currency the rest of the CBO work extends (byte-based exchange costs once per-column `avgByteSize` lands), so I'd rather introduce it once, here, than migrate costs later. ########## pinot-query-planner/src/main/java/org/apache/pinot/calcite/plan/PinotRelOptCost.java: ########## @@ -0,0 +1,316 @@ +/** + * 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.calcite.plan; + +import java.util.Objects; +import org.apache.calcite.plan.RelOptCost; +import org.apache.calcite.plan.RelOptCostFactory; + + +/// Pinot-specific implementation of [RelOptCost] for the multi-stage query engine (MSE). +/// +/// ### Ordering semantics (rows-dominated lexicographic) +/// Cost comparison uses a strict (rows, cpu, io) lexicographic order: +/// 1. Row count is the primary key — plans that process fewer rows are cheaper. +/// 1. CPU cost breaks ties when row counts are equal. +/// 1. IO cost breaks ties when both rows and CPU are equal. +/// +/// ### Deliberate difference from Calcite's `VolcanoCost` +/// Calcite's `VolcanoCost.isLe` compares *only* row count: two costs with the same +/// number of rows are considered equal regardless of their cpu or io values. This means +/// `VolcanoCost` has no way to break ties when row counts match, which can leave join-order +/// choices arbitrary. Review Comment: Fair observation — most of that surface is the `RelOptCost` interface contract rather than speculative API: `plus`/`minus`/`multiplyBy`/`divideBy`/`isEqWithEpsilon` and the `ZERO`/`TINY`/`HUGE`/`INFINITY` constants are what Calcite's planners and metadata handlers call (cost accumulation runs through `plus`, comparisons run against zero/infinity, and the `Factory` needs `makeZero`/`makeInfinite`/…). What this PR actively exercises is the factory wiring in `JoinReorderOptimizer` plus `isLe`/`plus` during reorder; the rest can't be trimmed without failing the interface. The bigger picture is exactly the follow-up you're guessing at: this class is the shared cost model the physical/distribution phase reuses, so the io/cpu components get real weights there instead of a second cost type appearing later. ########## pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java: ########## Review Comment: Good catch on the typos — applied in bcc247420e, with one more fix (the suggestion had a duplicated "in fact") and slightly different placement wording so it matches the code: the cost-based reorder runs *inside* this method, after the standard rule programs and before trait resolution. ########## pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java: ########## @@ -513,6 +524,15 @@ private RelNode optimize(RelRoot relRoot, PlannerContext plannerContext) { RelNode optimized = optPlanner.findBestExp(); listener.printRuleTimings(); listener.populateRuleTimings(); + // Scoped, gated, cost-based join-reordering phase. Runs after the logical Hep program and + // before the trait phase. Off by default; when disabled or when its eligibility gates fail + // it returns the plan unchanged. It never throws — see JoinReorderOptimizer.maybeReorder. + if (QueryOptionsUtils.isUseJoinReorder(plannerContext.getOptions(), + _envConfig.defaultUseJoinReorder())) { + int maxJoins = QueryOptionsUtils.getJoinReorderMaxJoins(plannerContext.getOptions(), + _envConfig.defaultJoinReorderMaxJoins()); + optimized = JoinReorderOptimizer.maybeReorder(optimized, maxJoins); + } RelOptPlanner traitPlanner = plannerContext.getRelTraitPlanner(); traitPlanner.setRoot(optimized); return traitPlanner.findBestExp(); Review Comment: Great question — this is the classic access-path × join-order interaction (System R solved them jointly for exactly this reason). Three Pinot-specific observations on why the split is safer here than in the textbook case: 1. **Scan choice in Pinot is a per-segment, server-side decision.** `FilterOperatorUtils` picks per segment, against that segment's actual indexes — and segments of one table genuinely differ (consuming vs. committed, index changes over time). The broker can't make that decision at plan time even in principle; it can only estimate an *average* scan cost. 2. **Join order is driven by cardinalities, and cardinalities don't depend on scan implementation.** How a filter is evaluated changes CPU, not how many rows come out. So reorder-then-scan loses optimality only where scan-cost differences are large enough to flip an order — and that class (a selective index making "probe a big table with few keys" cheap) is better modeled as costed **join strategies** (lookup join, dynamic-broadcast/filter semi-joins) than as leaf scan choice. That's where it's headed in the physical-phase design. 3. **The nearer-term win for scan choice needs no broker at all**: servers already hold their segments' stats locally, so `FilterOperatorUtils` could become smarter with local information; the broker's marginal contribution would be pushing down predicate *selectivity* estimates as query metadata. Both are good follow-ups independent of join order. If evidence ever shows the interaction really costs us plans, the escape hatch is an integrated search over both — but that's a planning-latency trade-off that deserves its own numbers first; it's discussed in the design doc for the physical phase. ########## 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, Review Comment: On the evolution question specifically: confidence is attached **per statistic**, not per table, so a future upsert estimator (like the periodic query above) can emit `ESTIMATED` for the tables/columns it has sampled and leave `LOW` elsewhere — different qualities per table fall out naturally. The enum is deliberately coarse because the planner contract is binary today (trustworthy enough for cost decisions, or treated as absent); if we ever need finer grading, adding tiers is SPI-compatible since consumers compare ordinally. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
