xiangfu0 commented on code in PR #19255:
URL: https://github.com/apache/pinot/pull/19255#discussion_r3791173156
##########
pinot-controller/src/main/java/org/apache/pinot/controller/util/ServerSegmentMetadataReader.java:
##########
@@ -179,10 +204,30 @@ public TableMetadataInfo
getAggregatedTableMetadataFromServer(String tableNameWi
? new ServerCompressionStatsReader(_executor,
_connectionManager).read(tableNameWithType,
serverToSegmentsMap, serverEndPoints, columns,
includeColumnCompressionStats, deadlineNanos) : null;
+ // Report the breakdown only if every server answered. `totalNumSegments`
only counts parsed responses, so a
+ // failed parse leaves no trace a caller could use to notice the shortfall
-- and a per-index-type total that
+ // silently omits a server is worse than no total at all.
+ Map<String, IndexSizeBreakdownInfo> indexSizeBreakdown = null;
+ if (indexSizeTotals.isEmpty()) {
+ LOGGER.debug("No server reported index size stats for table: {}",
tableNameWithType);
+ } else if (failedParses != 0) {
Review Comment:
**[P1] Do not publish a partial or mixed-version aggregate.** `failedParses`
misses HTTP failures/timeouts, which are absent from `_httpResponses`, and
successful old-server responses where `indexSizeBreakdown` is null. With RF=2,
one 100-byte response plus one timeout is published as 50 bytes, and one
contributing segment truncates to zero. Please require every requested endpoint
to return a parseable, capable response before publishing this field, or
aggregate one donor per logical segment. Add mixed old/new, non-2xx, and
timeout coverage.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java:
##########
@@ -465,9 +473,47 @@ public static ColumnMetadataImpl
fromPropertiesConfiguration(PropertiesConfigura
Column.getKeyFor(column,
Column.FORWARD_INDEX_DICTIONARY_ENCODED_UNCOMPRESSED_VALUE_SIZE_IN_BYTES),
UNAVAILABLE));
+ // Read the per-index on-disk sizes persisted at seal time and refreshed
opportunistically on reload, keyed by
+ // index type id -- see ColumnMetadata#getPersistedIndexSizesInBytes for
the exact staleness semantics. Kept
+ // separate from the packed sizes that [SegmentMetadataImpl] loads from
`v3/index_map`: those describe the live
+ // layout and require the segment payload; these are readable from
`metadata.properties` alone.
+ builder.setPersistedIndexSizesInBytes(readPersistedIndexSizes(config,
column));
+
return builder.build();
}
+ /// Collects `column.<column>.indexSizeInBytes.<indexTypeId>` entries for
`column`. Unparseable values are skipped:
+ /// these statistics are advisory and must never prevent a segment from
loading. Never null; empty when nothing was
+ /// persisted.
+ private static Map<String, Long>
readPersistedIndexSizes(PropertiesConfiguration config, String column) {
+ String keyPrefix = Column.getKeyFor(column, Column.INDEX_SIZE_IN_BYTES) +
".";
+ Map<String, Long> indexSizes = null;
+ for (String key : CommonsConfigurationUtils.getKeys(config)) {
Review Comment:
**[P2] Avoid rescanning all properties for every column.**
`fromPropertiesConfiguration()` calls this once per physical column, while
`getKeys(config)` materializes and scans every metadata key each time. That
makes every non-empty segment load roughly `O(columns x properties)`, even when
this opt-in data is absent. Please collect/group these entries once per
segment, or read one exact per-column property and pass the result down.
##########
pinot-server/src/main/java/org/apache/pinot/server/api/resources/TablesResource.java:
##########
@@ -285,6 +292,28 @@ public String getSegmentMetadata(
columnIndexSizesMap.put(column, columnIndexSizes);
}
}
+
+ if (indexSizeTotals != null) {
+ // Deliberately iterates every column of the segment, not
`columnSet`: indexSizeBreakdown is a
+ // table-level per-index-type total and does not honour the
`columns=` filter, unlike columnIndexSizeMap
+ // above. Sizes come from the keys persisted at seal time, so a
segment built without
+ // indexSizeStatsEnabled contributes nothing and is not counted.
+ for (String column : allSegmentColumns) {
+ ColumnMetadata metadata =
segmentMetadata.getColumnMetadataMap().get(column);
+ if (metadata == null) {
+ continue;
+ }
+ for (Map.Entry<String, Long> entry :
metadata.getPersistedIndexSizesInBytes().entrySet()) {
+ long size = entry.getValue();
+ if (size < 0) {
+ continue;
+ }
+ long[] total =
indexSizeTotals.computeIfAbsent(entry.getKey(), k -> new long[2]);
+ total[0] += size;
+ total[1]++;
Review Comment:
**[P1] Count segments, not `(column, indexType)` entries.** This increment
is inside the per-column loop, so one segment with ten forward-indexed columns
reports `segmentsWithStats=10`. Please sum bytes per type across the segment
first, then increment each type once. A populated multi-column endpoint test
should assert the exact count.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/index/loader/SegmentPreProcessor.java:
##########
@@ -177,17 +183,260 @@ public void process(@Nullable
SegmentOperationsThrottlerSet segmentOperationsThr
// Startree creation will load the segment again, so we need to close and
re-open the segment writer to make sure
// that the other required indices (e.g. forward index) are up-to-date.
+ IndexPresenceSnapshot indexPresenceSnapshot = null;
try (SegmentDirectory.Writer segmentWriter =
_segmentDirectory.createWriter()) {
if (processStarTrees(indexDir, segmentOperationsThrottlerSet)) {
_segmentDirectory.reloadMetadata();
segmentWriter.save();
}
// Create/modify/remove multi-col text index if required.
if (processMultiColTextIndex(indexDir, segmentWriter,
segmentOperationsThrottlerSet)) {
- // NOTE: When adding new steps after this, un-comment the next line.
- //_segmentDirectory.reloadMetadata();
+ _segmentDirectory.reloadMetadata();
segmentWriter.save();
}
+
+ // Snapshot which (column, indexType) pairs exist right now, straight
from this still-open writer -- not from
+ // a later independent metadata read -- so refreshPersistedIndexSizes()
reconciles against the exact on-disk
+ // layout this reload just produced. See its javadoc for why a live
snapshot matters here.
+ if (_tableConfig.getIndexingConfig().isIndexSizeStatsEnabled()) {
+ try {
+ List<IndexType<?, ?, ?>> allIndexTypes =
IndexService.getInstance().getAllIndexes();
+ IndexPresenceSnapshot snapshot = snapshotIndexTypeIds(segmentWriter,
+
_segmentDirectory.getSegmentMetadata().getColumnMetadataMap().keySet(),
allIndexTypes);
+ if (snapshot.getColumnToIndexTypeIds().isEmpty()) {
+ // A non-empty segment always has a forward index or dictionary on
every column, so an empty snapshot
+ // means the backing SegmentDirectory answered "no indexes
anywhere" rather than reporting a genuine
+ // state -- e.g. SegmentLocalFSDirectory#getColumnsWithIndex
returns Set.of() for every type while its
+ // column-index directory is not loaded. Treat it the same as a
snapshot failure: skip the refresh
+ // rather than let a spurious empty answer clear every persisted
size below.
+ LOGGER.warn("Post-reload index snapshot for segment: {} was
unexpectedly empty; skipping index size "
+ + "stats refresh for this reload", segmentName);
+ } else {
+ // Only publish a fully-validated snapshot, so that a failure
anywhere above -- including inside this
+ // same try block, e.g. the LOGGER.warn call above throwing --
always leaves indexPresenceSnapshot null
+ // and the refresh skipped below, with no dependence on how far
the try body got before failing.
+ indexPresenceSnapshot = snapshot;
+ }
+ } catch (Exception e) {
+ // Advisory stats must never fail a segment load: skip the size
refresh for this reload entirely rather
+ // than let the whole process() call fail. Per-index-type probe
failures are handled inside
+ // snapshotIndexTypeIds() itself and do not reach this catch; this
remains as a backstop for anything
+ // else unexpected (e.g. IndexService.getAllIndexes() itself
misbehaving).
+ LOGGER.warn("Failed to snapshot post-reload index sizes for segment:
{}; skipping index size stats "
+ + "refresh for this reload", segmentName, e);
+ }
+ }
+ }
+
+ // Every index handler has finished, so the on-disk layout is final:
refresh the persisted per-index sizes.
+ // This is opportunistic only: it rides along with whatever reload just
ran for some other reason, and never
+ // itself decides that a reload is needed. See needProcess() javadoc for
why index size stats are excluded from
+ // that decision entirely.
+ if (indexPresenceSnapshot != null) {
+ refreshPersistedIndexSizes(indexDir, indexPresenceSnapshot);
+ }
+ }
+
+ /// Result of [#snapshotIndexTypeIds]: for every column with at least one
index, the set of [IndexType#getId]
+ /// values present on it right now, plus the set of index type ids that were
actually, successfully probed while
+ /// building that map. The two are not redundant: a probe failure for one
index type (see
+ /// [#snapshotIndexTypeIds]'s javadoc) makes that type's presence unknown
for this reload, not absent, so
+ /// [#refreshPersistedIndexSizes] must be able to tell "probed and confirmed
absent" apart from "never probed."
+ private static final class IndexPresenceSnapshot {
+ private final Map<String, Set<String>> _columnToIndexTypeIds;
+ private final Set<String> _probedIndexTypeIds;
+
+ private IndexPresenceSnapshot(Map<String, Set<String>>
columnToIndexTypeIds, Set<String> probedIndexTypeIds) {
+ _columnToIndexTypeIds = columnToIndexTypeIds;
+ _probedIndexTypeIds = probedIndexTypeIds;
+ }
+
+ private Map<String, Set<String>> getColumnToIndexTypeIds() {
+ return _columnToIndexTypeIds;
+ }
+
+ private Set<String> getPresentIndexTypeIds(String column) {
+ return _columnToIndexTypeIds.getOrDefault(column, Set.of());
+ }
+
+ private boolean wasProbed(String indexTypeId) {
+ return _probedIndexTypeIds.contains(indexTypeId);
+ }
+ }
+
+ /// Returns, for every column in `columnsToInclude` that currently has at
least one index, the set of
+ /// [IndexType#getId] values present on it right now. Always read from a
live [SegmentDirectory.Reader] (a
+ /// [SegmentDirectory.Writer] qualifies too), never from an independent
`SegmentMetadataImpl` re-read of
+ /// `metadata.properties` or `v3/index_map` -- see
[#refreshPersistedIndexSizes] for why that distinction matters.
+ ///
+ /// Loops index types in the outer loop and calls
`SegmentDirectory#getColumnsWithIndex` per type rather than
+ /// looping columns and calling `hasIndexFor` per (column, indexType) pair,
but this does not make the call cheap:
+ /// depending on the backing store, `getColumnsWithIndex` itself may scan
every column for every call
+ /// (`FilePerIndexDirectory`) or every entry for every call
(`SingleFileIndexDirectory`), so total cost still scales
+ /// with `allIndexTypes.size()` times the backing store's per-call cost, not
with the number of present indexes.
+ ///
+ /// A per-index-type probe failure (e.g.
`FilePerIndexDirectory#getColumnsWithIndex` throwing because a registered
+ /// [IndexType#getFileExtensions] returns an empty list) is caught here and
only drops that one index type from the
+ /// snapshot; it does not fail the whole call. This matters because that
failure mode recurs on every future reload
+ /// for the same segment and index type, so treating it as "snapshot failed,
skip the whole refresh" would
+ /// permanently stop refreshing every other index type on this segment too,
not just this one.
+ private static IndexPresenceSnapshot
snapshotIndexTypeIds(SegmentDirectory.Reader reader,
+ Set<String> columnsToInclude, List<IndexType<?, ?, ?>> allIndexTypes) {
+ Map<String, Set<String>> columnToIndexTypeIds = new HashMap<>();
+ Set<String> probedIndexTypeIds = new HashSet<>();
+ SegmentDirectory segmentDirectory = reader.toSegmentDirectory();
+ for (IndexType<?, ?, ?> indexType : allIndexTypes) {
+ Set<String> columnsWithIndex;
+ try {
+ columnsWithIndex = segmentDirectory.getColumnsWithIndex(indexType);
+ } catch (Exception e) {
+ LOGGER.warn("Failed to probe index type: {} while snapshotting
post-reload index presence; treating its "
+ + "presence as unknown for this reload rather than absent",
indexType.getId(), e);
+ continue;
+ }
+ probedIndexTypeIds.add(indexType.getId());
+ for (String column : columnsWithIndex) {
+ if (columnsToInclude.contains(column)) {
+ columnToIndexTypeIds.computeIfAbsent(column, c -> new
HashSet<>()).add(indexType.getId());
+ }
+ }
+ }
+ return new IndexPresenceSnapshot(columnToIndexTypeIds, probedIndexTypeIds);
+ }
+
+ /// Updates the `column.<column>.indexSizeInBytes.<indexTypeId>` entries in
`metadata.properties` to match
+ /// `indexPresenceSnapshot`, the live post-reload index presence for every
column, and leaves every other entry
+ /// untouched.
+ ///
+ /// Without this the values would stay a build-time snapshot: reload can
add, drop or re-compress an index, and a
+ /// stale size is indistinguishable from a current one. This is the only
hook, deliberately: index sizes span every
+ /// index type, so updating them per handler -- as `compressionStatsEnabled`
does in `ForwardIndexHandler`, where a
+ /// single index is involved -- would require every future handler to
remember to participate.
+ ///
+ /// For every column and every currently-registered [IndexType]:
+ /// - Present (per `indexPresenceSnapshot`): (re)sized the same way segment
creation would size it -- from the
+ /// packed [ColumnMetadata#getIndexSize] position if there is one, else
from the index's own file/directory --
+ /// which is reliable here specifically because this reload just wrote the
current layout, locally, moments ago.
+ /// The size is only written if it differs from what is already persisted,
so a reload that left an index
+ /// untouched writes nothing. Refreshing every present index rather than
only newly-added ones is deliberate: a
+ /// handler can remove and recreate an index of the same type within one
reload -- e.g.
+ /// `LegacyRawValueInvertedIndexCleanup` dropping a legacy-format inverted
index for `InvertedIndexHandler` to
+ /// rebuild, or `ForwardIndexHandler` changing a raw column's compression
codec -- which leaves presence
+ /// unchanged across the reload while the actual size changes. Presence
alone cannot see that; comparing the
+ /// freshly computed size against the persisted one can.
+ /// - Absent (per `indexPresenceSnapshot`) among the index types that were
actually, successfully probed while
+ /// building that snapshot: its persisted key, if any, is cleared. This
intentionally does not require having
+ /// observed the index as present on some earlier reload: a successfully
probed index type reports every column
+ /// with at least one index, live, right now, so that index type being
absent from a column's present set is
+ /// confirmed absent for that column, not merely unobserved. Clearing
unconditionally on that basis is what
+ /// reconciles a phantom size left behind by, for example, a reload that
dropped an index while
+ /// `indexSizeStatsEnabled` was off (which skips this method entirely)
followed by a later reload with the flag
+ /// back on. An index type that failed to probe (see
[#snapshotIndexTypeIds]) is excluded from this clearing:
+ /// its absence from a column's present set means "unknown," not
"confirmed absent," so its persisted entry, if
+ /// any, is left alone -- the same "leave unchanged" treatment as an
unmeasurable present index below. This also
+ /// covers an index type this node has no plugin for at all: such an id
can never appear as probed, since probing
+ /// only iterates locally-registered [IndexType]s, so a size persisted by
a node or version with a plugin this
+ /// node lacks survives reload here rather than being wiped by a node that
cannot even see it exists.
+ /// No tier logic anywhere here: the same rule applies to every segment
format and every storage tier.
+ ///
+ /// Only called by [#process] with a non-null, already-validated
`indexPresenceSnapshot` (non-empty, and produced
+ /// without the snapshotting step itself throwing), so this method re-checks
neither `indexSizeStatsEnabled` nor
+ /// emptiness. Failures
+ /// are logged and swallowed: these statistics are advisory and must never
fail a segment load, and -- because
+ /// nothing outside this method ever asks "are the persisted sizes still
accurate," see [#needProcess] -- must
+ /// never force one either. Refreshing only happens as a side effect of a
reload that some other check already
+ /// decided was needed.
+ ///
+ /// Runs after `process()`'s own [SegmentDirectory.Writer] has already
closed, and reads `indexDir` directly off
+ /// disk rather than through the writer, so it relies on the same invariant
every other unguarded step of
+ /// `process()` does: the caller holds the per-segment lock (see
`BaseTableDataManager`'s segment locks) for the
+ /// duration of this call, so nothing else concurrently mutates this segment
directory.
+ private void refreshPersistedIndexSizes(File indexDir, IndexPresenceSnapshot
indexPresenceSnapshot) {
+ try {
+ // Read the metadata fresh off disk rather than using
_segmentDirectory.getSegmentMetadata(): that returns a
+ // cached instance describing the layout as it was before the handlers
ran. Only used below to size present
+ // indexes, never to decide what is present -- that comes from
indexPresenceSnapshot.
+ SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir);
+ if (segmentMetadata.getTotalDocs() == 0) {
+ return;
+ }
+ IndexService indexService = IndexService.getInstance();
+ PropertiesConfiguration properties =
SegmentMetadataUtils.getPropertiesConfiguration(indexDir);
+ File segmentContentDir =
SegmentDirectoryPaths.findSegmentDirectory(indexDir);
+ Map<String, ColumnMetadata> columnMetadataMap =
segmentMetadata.getColumnMetadataMap();
+
+ boolean propertiesChanged = false;
+ for (Map.Entry<String, ColumnMetadata> columnEntry :
columnMetadataMap.entrySet()) {
+ String column = columnEntry.getKey();
+ ColumnMetadata columnMetadata = columnEntry.getValue();
+ Set<String> presentIndexTypeIds =
indexPresenceSnapshot.getPresentIndexTypeIds(column);
+
+ // Clear the persisted size for every previously-persisted index type
confirmed absent on this column --
+ // successfully probed but not present -- regardless of whether this
particular reload is the one that
+ // removed it; see the javadoc above for why this is safe and why it
is also what reconciles a
+ // flag-off/flag-on toggle. An index type that was never successfully
probed (unregistered on this node, or a
+ // probe failure) is left untouched instead: see the javadoc above.
Walking only the keys already on disk,
+ // rather than every registered index type, keeps this proportional to
what is actually persisted.
+ for (String indexTypeId :
columnMetadata.getPersistedIndexSizesInBytes().keySet()) {
+ if (indexPresenceSnapshot.wasProbed(indexTypeId) &&
!presentIndexTypeIds.contains(indexTypeId)) {
Review Comment:
**[P1] Preserve untouched entries when the tier manifest is partial.**
`wasProbed()` only means `getColumnsWithIndex()` returned; it does not prove
the result is complete. A remote-tier directory can expose a non-empty
materialized subset, so an untouched `(column, indexType)` is treated as absent
and its persisted size is deleted here. Please clear only indexes explicitly
removed by this reload, or require an explicit complete-manifest signal before
clearing. Add a partial (not all-empty) manifest regression test.
##########
pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/SegmentSizeInfo.java:
##########
@@ -66,13 +87,27 @@ public SegmentSizeInfo(@JsonProperty("segmentName") String
segmentName,
@JsonProperty("compressionStatsForwardIndexAndDictionaryStorageSizeInBytes")
@Nullable
Long compressionStatsForwardIndexAndDictionaryStorageSizeInBytes,
@JsonProperty("columnCompressionStats") @Nullable Map<String,
ColumnCompressionStatsInfo>
- columnCompressionStats) {
+ columnCompressionStats,
+ @JsonProperty("indexSizeInBytes") @Nullable Map<String, Long>
indexSizeInBytes) {
Review Comment:
**[P1] Restore the exact boxed five-argument constructor.** Adding this
parameter removes the public `(String, long, Long, Long, Map)` descriptor.
Existing binaries can fail with `NoSuchMethodError`; the primitive overload is
not equivalent and can auto-unbox nullable callers to an NPE. Please retain the
old overload and delegate here with `indexSizeInBytes = null`.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/index/metadata/ColumnMetadataImpl.java:
##########
@@ -465,9 +473,47 @@ public static ColumnMetadataImpl
fromPropertiesConfiguration(PropertiesConfigura
Column.getKeyFor(column,
Column.FORWARD_INDEX_DICTIONARY_ENCODED_UNCOMPRESSED_VALUE_SIZE_IN_BYTES),
UNAVAILABLE));
+ // Read the per-index on-disk sizes persisted at seal time and refreshed
opportunistically on reload, keyed by
+ // index type id -- see ColumnMetadata#getPersistedIndexSizesInBytes for
the exact staleness semantics. Kept
+ // separate from the packed sizes that [SegmentMetadataImpl] loads from
`v3/index_map`: those describe the live
+ // layout and require the segment payload; these are readable from
`metadata.properties` alone.
+ builder.setPersistedIndexSizesInBytes(readPersistedIndexSizes(config,
column));
+
return builder.build();
}
+ /// Collects `column.<column>.indexSizeInBytes.<indexTypeId>` entries for
`column`. Unparseable values are skipped:
+ /// these statistics are advisory and must never prevent a segment from
loading. Never null; empty when nothing was
+ /// persisted.
+ private static Map<String, Long>
readPersistedIndexSizes(PropertiesConfiguration config, String column) {
+ String keyPrefix = Column.getKeyFor(column, Column.INDEX_SIZE_IN_BYTES) +
".";
+ Map<String, Long> indexSizes = null;
+ for (String key : CommonsConfigurationUtils.getKeys(config)) {
+ if (!key.startsWith(keyPrefix)) {
Review Comment:
**[P1] This prefix is ambiguous for dotted column names.** With columns
`foo` and `foo.indexSizeInBytes.bar`, the ordinary
`column.foo.indexSizeInBytes.bar.cardinality` property for the latter column
matches here and becomes a fake size entry `bar.cardinality`, which then
pollutes the API totals even when size collection was never enabled. Please use
an unambiguous representation/ownership check and add this overlapping-name
regression case.
##########
pinot-common/src/main/java/org/apache/pinot/common/restlet/resources/TableMetadataInfo.java:
##########
@@ -58,6 +64,7 @@ public TableMetadataInfo(@JsonProperty("tableName") String
tableName,
Map<Integer, Map<String, Long>> partitionToServerPrimaryKeyCountMap,
@JsonProperty("columnCompressionStats") @Nullable
List<ColumnCompressionStatsInfo> columnCompressionStats,
+ @JsonProperty("indexSizeBreakdown") @Nullable Map<String,
IndexSizeBreakdownInfo> indexSizeBreakdown,
Review Comment:
**[P1] Preserve the existing eleven-argument constructor descriptor.**
Adding this parameter removes the public overload ending in
`(List<ColumnCompressionStatsInfo>, CompressionStatsSummary)`, so old source no
longer compiles and existing binaries can fail with `NoSuchMethodError`. Please
restore that exact overload and delegate with `indexSizeBreakdown = null`.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/segment/creator/impl/BaseSegmentCreator.java:
##########
@@ -410,7 +412,10 @@ private Object calculateRawValueForTextIndex(boolean
dictEnabledColumn, FieldInd
}
/// Writes segment metadata to disk.
- protected void writeMetadata()
+ /// Writes `metadata.properties`. `indexSizes` maps column to
[IndexType#getId()] to on-disk size in bytes and is
+ /// empty unless `indexSizeStatsEnabled` is set; when populated, the sizes
are written as
+ /// `column.<column>.indexSizeInBytes.<indexTypeId>` in this same single
pass.
+ protected void writeMetadata(Map<String, Map<String, Long>> indexSizes)
Review Comment:
**[P1] Preserve the protected zero-argument extension hook.** Replacing
`writeMetadata()` means existing overrides no longer compile or dispatch, and
compiled subclasses calling `super.writeMetadata()` can fail with
`NoSuchMethodError`. Please retain and continue invoking the no-argument
method; its base implementation can delegate to a private map-aware helper.
--
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]
