diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java index d1d1f3bb3062..b2d5cf247297 100644 --- a/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java +++ b/pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManager.java @@ -2354,11 +2354,19 @@ private void handleMetadataUpload(String rawTableName, String segmentName, Segme if (currentMetadata.getStatus() == Status.COMMITTING) { LOGGER.info("Updating ZK metadata for committing segment: {}", segmentName); currentMetadata.setSimpleFields(uploadedMetadata.getSimpleFields()); - } else if (currentMetadata.getCrc() != uploadedMetadata.getCrc()) { - LOGGER.info("Updating CRC in ZK metadata for segment: {} from: {} to: {}", segmentName, currentMetadata.getCrc(), - uploadedMetadata.getCrc()); - currentMetadata.setCrc(uploadedMetadata.getCrc()); - currentMetadata.setDataCrc(uploadedMetadata.getDataCrc()); + } else { + if (currentMetadata.getCrc() != uploadedMetadata.getCrc()) { + LOGGER.info("Updating CRC in ZK metadata for segment: {} from: {} to: {}", segmentName, + currentMetadata.getCrc(), uploadedMetadata.getCrc()); + currentMetadata.setCrc(uploadedMetadata.getCrc()); + currentMetadata.setDataCrc(uploadedMetadata.getDataCrc()); + } + long uploadedSizeInBytes = uploadedMetadata.getSizeInBytes(); + if (uploadedSizeInBytes >= 0 && currentMetadata.getSizeInBytes() != uploadedSizeInBytes) { + LOGGER.info("Updating size in ZK metadata for segment: {} from: {} to: {}", segmentName, + currentMetadata.getSizeInBytes(), uploadedSizeInBytes); + currentMetadata.setSizeInBytes(uploadedSizeInBytes); + } } // Merge the custom map from the server into the current metadata. The server merges segment-file customMap entries // into the existing ZK customMap; without this merge those entries would be dropped when we persist diff --git a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java index 45e85b57e640..983201ac1f45 100644 --- a/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java +++ b/pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/realtime/PinotLLCRealtimeSegmentManagerTest.java @@ -131,6 +131,7 @@ public class PinotLLCRealtimeSegmentManagerTest { private static final String REALTIME_TABLE_NAME = TableNameBuilder.REALTIME.tableNameWithType(RAW_TABLE_NAME); private static final String SEGMENT_LOCATION_PREFIX = "http://control_vip/segments/"; private static final int SEGMENT_SIZE_IN_BYTES = 100000000; + private static final int FINALIZED_SEGMENT_SIZE_IN_BYTES = 25000000; private static final long CURRENT_TIME_MS = System.currentTimeMillis(); private static final Random RANDOM = new Random(CURRENT_TIME_MS); @@ -1823,11 +1824,13 @@ public void testUploadCommittedSegment() Map existingCustomMap = new HashMap<>(); existingCustomMap.put("controllerKey", "controllerValue"); segmentsZKMetadata.get(0).setCustomMap(existingCustomMap); + segmentsZKMetadata.get(0).setSizeInBytes(SEGMENT_SIZE_IN_BYTES); SegmentZKMetadata segmentZKMetadataCopy = new SegmentZKMetadata(new ZNRecord(segmentsZKMetadata.get(0).toZNRecord())); segmentZKMetadataCopy.setDownloadUrl(tempSegmentFileLocation.getPath()); + segmentZKMetadataCopy.setSizeInBytes(FINALIZED_SEGMENT_SIZE_IN_BYTES); // Simulate the server-side merge of segment-file customMap entries into the uploaded ZK metadata. The controller // must propagate these entries to the persisted ZK metadata. Map uploadedCustomMap = new HashMap<>(); @@ -1891,6 +1894,9 @@ public void testUploadCommittedSegment() assertNotNull(persistedCustomMap); assertEquals(persistedCustomMap.get("segmentFileKey"), "segmentFileValue"); assertEquals(persistedCustomMap.get("controllerKey"), "controllerValue"); + assertEquals( + segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, segmentNames.get(0), null).getSizeInBytes(), + FINALIZED_SEGMENT_SIZE_IN_BYTES); assertEquals(segmentManager.getSegmentZKMetadata(REALTIME_TABLE_NAME, segmentNames.get(1), null).getDownloadUrl(), METADATA_URI_FOR_PEER_DOWNLOAD); diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java index 47695c74ec28..17f2a20f74c5 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java @@ -59,6 +59,7 @@ import org.apache.pinot.segment.local.data.manager.SegmentDataManager; import org.apache.pinot.segment.local.dedup.DedupContext; import org.apache.pinot.segment.local.dedup.PartitionDedupMetadataManager; +import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader; import org.apache.pinot.segment.local.indexsegment.mutable.MutableSegmentImpl; import org.apache.pinot.segment.local.io.writer.impl.DirectMemoryManager; import org.apache.pinot.segment.local.io.writer.impl.MmapMemoryManager; @@ -1286,7 +1287,18 @@ protected SegmentBuildDescriptor buildSegmentInternal(boolean forCommit) FileUtils.deleteQuietly(tempSegmentFolder); } - long segmentSizeBytes = FileUtils.sizeOfDirectory(indexDir); + try { + IndexLoadingConfig latestIndexLoadingConfig = _realtimeTableDataManager.fetchIndexLoadingConfig(); + ImmutableSegmentLoader.preprocess(indexDir, latestIndexLoadingConfig, + _realtimeTableDataManager.getSegmentOperationsThrottlerSet(), _segmentZKMetadata); + } catch (Exception e) { + String errorMessage = "Could not finalize segment with the latest index configuration"; + reportSegmentBuildFailure(errorMessage, e); + throw new SegmentBuildFailureException(errorMessage, e); + } + + long segmentSizeBytes = + FileUtils.sizeOfDirectory(SegmentDirectoryPaths.findSegmentDirectory(indexDir)); _serverMetrics.setValueOfTableGauge(_clientId, ServerGauge.LAST_REALTIME_SEGMENT_CREATION_DURATION_SECONDS, TimeUnit.MILLISECONDS.toSeconds(buildTimeMillis)); _serverMetrics.setValueOfTableGauge(_clientId, ServerGauge.LAST_REALTIME_SEGMENT_CREATION_WAIT_TIME_SECONDS, diff --git a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java index bb65dc48dbb5..b2f79ef4ff01 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java @@ -816,8 +816,9 @@ public void replaceConsumingSegment(String segmentName, @Nullable SegmentZKMetad File indexDir = new File(_indexDir, segmentName); // Get a new index loading config with latest table config and schema to load the segment IndexLoadingConfig indexLoadingConfig = fetchIndexLoadingConfig(); - ImmutableSegment immutableSegment = - ImmutableSegmentLoader.load(indexDir, indexLoadingConfig, _segmentOperationsThrottlerSet, zkMetadata); + boolean needPreprocess = ImmutableSegmentLoader.needPreprocess(indexDir, indexLoadingConfig, zkMetadata); + ImmutableSegment immutableSegment = ImmutableSegmentLoader.load(indexDir, indexLoadingConfig, needPreprocess, + _segmentOperationsThrottlerSet, zkMetadata); addSegment(immutableSegment, zkMetadata); _ingestionDelayTracker.markPartitionForVerification(segmentName); diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java index 95a64108511e..8f3815fc7db5 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java @@ -48,6 +48,8 @@ import org.apache.pinot.core.realtime.impl.fakestream.FakeStreamMessageDecoder; import org.apache.pinot.segment.local.data.manager.SegmentDataManager; import org.apache.pinot.segment.local.data.manager.TableDataManager; +import org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader; +import org.apache.pinot.segment.local.indexsegment.mutable.MutableSegmentImpl; import org.apache.pinot.segment.local.realtime.impl.RealtimeSegmentStatsHistory; import org.apache.pinot.segment.local.segment.creator.Fixtures; import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl; @@ -55,6 +57,7 @@ import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader; import org.apache.pinot.segment.local.utils.SegmentLocks; import org.apache.pinot.segment.local.utils.ServerReloadJobStatusCache; +import org.apache.pinot.segment.spi.ImmutableSegment; import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; import org.apache.pinot.segment.spi.index.metadata.SegmentMetadataImpl; import org.apache.pinot.spi.config.instance.InstanceDataManagerConfig; @@ -789,6 +792,44 @@ public void testIsLocalSegmentCrcMatchingZkTreatsUnreadableSegmentAsMismatch() } } + @Test + public void testSegmentBuildSizeUsesLatestIndexConfiguration() + throws Exception { + TableConfig buildConfig = createTableConfig(); + buildConfig.getIndexingConfig().setSegmentFormatVersion("v3"); + buildConfig.getIndexingConfig().setInvertedIndexColumns(List.of("d")); + + try (FakeRealtimeSegmentDataManager segmentDataManager = + createFakeSegmentManager(false, new TimeSupplier(), null, null, buildConfig)) { + Schema schema = Fixtures.createSchema(); + TableConfig latestConfig = createTableConfig(); + latestConfig.getIndexingConfig().setSegmentFormatVersion("v3"); + IndexLoadingConfig latestIndexLoadingConfig = + new IndexLoadingConfig(FakeRealtimeSegmentDataManager.makeInstanceDataManagerConfig(), latestConfig, schema); + when(segmentDataManager._tableDataManager.fetchIndexLoadingConfig()).thenReturn(latestIndexLoadingConfig); + + for (int i = 0; i < 1_000; i++) { + GenericRow row = Fixtures.createSingleRow(i); + row.putValue("m", (long) i); + row.putValue("d", "value_" + i % 10); + row.putValue("minutesSinceEpoch", (long) i); + segmentDataManager.index(row); + } + + RealtimeSegmentDataManager.SegmentBuildDescriptor descriptor = + segmentDataManager.invokeRealBuildSegment(false); + File indexDir = new File(new File(TEMP_DIR, REALTIME_TABLE_NAME), SEGMENT_NAME_STR); + Assert.assertFalse(ImmutableSegmentLoader.needPreprocess(indexDir, latestIndexLoadingConfig, null)); + ImmutableSegment immutableSegment = ImmutableSegmentLoader.load(indexDir, latestIndexLoadingConfig, false); + try { + Assert.assertEquals(descriptor.getSegmentSizeBytes(), immutableSegment.getSegmentSizeBytes()); + Assert.assertNull(immutableSegment.getInvertedIndex("d")); + } finally { + immutableSegment.destroy(); + } + } + } + private long buildRealSegmentAndGetCrc(File resourceDir, String segmentName) throws Exception { Schema schema = Fixtures.createSchema(); @@ -1454,6 +1495,18 @@ public SegmentBuildDescriptor invokeBuildForCommit(long leaseTime) return getSegmentBuildDescriptor(); } + public SegmentBuildDescriptor invokeRealBuildSegment(boolean forCommit) + throws SegmentBuildFailureException { + return super.buildSegmentInternal(forCommit); + } + + public void index(GenericRow row) + throws Exception { + Field field = RealtimeSegmentDataManager.class.getDeclaredField("_realtimeSegment"); + field.setAccessible(true); + ((MutableSegmentImpl) field.get(this)).index(row, null); + } + public boolean invokeCommit() throws Exception { return super.commitSegment("dummyUrl"); diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java index 0e16f2f4925c..00f3ff343503 100644 --- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java +++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/indexsegment/immutable/ImmutableSegmentLoader.java @@ -289,6 +289,28 @@ public static boolean needPreprocess(SegmentDirectory segmentDirectory, IndexLoa return new SegmentPreProcessor(segmentDirectory, indexLoadingConfig).needProcess(); } + /// Checks whether a local segment directory needs preprocessing for the supplied index loading configuration. + /// + /// @param indexDir local segment index directory + /// @param indexLoadingConfig index loading configuration to check against + /// @param zkMetadata segment metadata used to resolve segment custom configuration, or `null` + /// @return `true` when loading the segment requires preprocessing + public static boolean needPreprocess(File indexDir, IndexLoadingConfig indexLoadingConfig, + @Nullable SegmentZKMetadata zkMetadata) + throws Exception { + Preconditions.checkArgument(indexDir.isDirectory(), "Index directory: %s does not exist or is not a directory", + indexDir); + + SegmentMetadataImpl segmentMetadata = new SegmentMetadataImpl(indexDir); + if (segmentMetadata.getTotalDocs() == 0) { + return false; + } + try (SegmentDirectory segmentDirectory = loadSegmentDirectoryForPreprocess(indexDir, segmentMetadata.getName(), + segmentMetadata.getCrc(), indexLoadingConfig, zkMetadata)) { + return needPreprocess(segmentDirectory, indexLoadingConfig); + } + } + private static boolean needConvertSegmentFormat(IndexLoadingConfig indexLoadingConfig, SegmentMetadataImpl segmentMetadata) { SegmentVersion segmentVersionToLoad = indexLoadingConfig.getSegmentVersion(); @@ -322,6 +344,16 @@ private static void preprocessSegment(File indexDir, String segmentName, String IndexLoadingConfig indexLoadingConfig, @Nullable SegmentOperationsThrottlerSet segmentOperationsThrottlerSet, SegmentZKMetadata zkMetadata) throws Exception { + SegmentDirectory segmentDirectory = + loadSegmentDirectoryForPreprocess(indexDir, segmentName, segmentCrc, indexLoadingConfig, zkMetadata); + try (SegmentPreProcessor preProcessor = new SegmentPreProcessor(segmentDirectory, indexLoadingConfig)) { + preProcessor.process(segmentOperationsThrottlerSet); + } + } + + private static SegmentDirectory loadSegmentDirectoryForPreprocess(File indexDir, String segmentName, + String segmentCrc, IndexLoadingConfig indexLoadingConfig, @Nullable SegmentZKMetadata zkMetadata) + throws Exception { SegmentDirectoryLoaderContext segmentLoaderContext = new SegmentDirectoryLoaderContext.Builder() .setReadMode(indexLoadingConfig.getReadMode()) .setTableConfig(indexLoadingConfig.getTableConfig()) @@ -331,10 +363,7 @@ private static void preprocessSegment(File indexDir, String segmentName, String .setSegmentCrc(segmentCrc) .setSegmentCustomConfigs(zkMetadata != null ? zkMetadata.getCustomMap() : Map.of()) .build(); - SegmentDirectory segmentDirectory = - SegmentDirectoryLoaderRegistry.getDefaultSegmentDirectoryLoader().load(indexDir.toURI(), segmentLoaderContext); - try (SegmentPreProcessor preProcessor = new SegmentPreProcessor(segmentDirectory, indexLoadingConfig)) { - preProcessor.process(segmentOperationsThrottlerSet); - } + return SegmentDirectoryLoaderRegistry.getDefaultSegmentDirectoryLoader().load(indexDir.toURI(), + segmentLoaderContext); } } diff --git a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/LoaderTest.java b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/LoaderTest.java index 06aa22e233d0..17cf48ad622f 100644 --- a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/LoaderTest.java +++ b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/segment/index/loader/LoaderTest.java @@ -163,6 +163,9 @@ public void testIfNeedConvertSegmentFormat() // The newly generated segment is consistent with table config and schema, thus // in follow checks, whether it needs reprocess or not depends on segment format. + assertFalse(ImmutableSegmentLoader.needPreprocess(_indexDir, new IndexLoadingConfig(), null)); + assertFalse(ImmutableSegmentLoader.needPreprocess(_indexDir, _v1IndexLoadingConfig, null)); + assertTrue(ImmutableSegmentLoader.needPreprocess(_indexDir, _v3IndexLoadingConfig, null)); try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(_indexDir, ReadMode.mmap)) { // The segmentVersionToLoad is null, not leading to reprocess. assertFalse(ImmutableSegmentLoader.needPreprocess(segmentDirectory, new IndexLoadingConfig())); @@ -177,6 +180,7 @@ public void testIfNeedConvertSegmentFormat() // Need to reset `segmentDirectory` to point to the correct index directory after the above load since the path // changes + assertFalse(ImmutableSegmentLoader.needPreprocess(_indexDir, _v3IndexLoadingConfig, null)); try (SegmentDirectory segmentDirectory = new SegmentLocalFSDirectory(_indexDir, ReadMode.mmap)) { assertFalse(ImmutableSegmentLoader.needPreprocess(segmentDirectory, _v3IndexLoadingConfig)); } diff --git a/pinot-tools/src/main/resources/examples/compression-stats/README.md b/pinot-tools/src/main/resources/examples/compression-stats/README.md index 5b2cac896f66..f667b4f75abd 100644 --- a/pinot-tools/src/main/resources/examples/compression-stats/README.md +++ b/pinot-tools/src/main/resources/examples/compression-stats/README.md @@ -35,6 +35,36 @@ resolved chunk-compression type. Dictionary-encoded columns record uncompressed `forwardIndexAndDictionaryStorageSizeInBytes` includes both the dictionary and forward-index files. Columns without a forward index and old segments without value-size metadata are excluded from compression ratios. +## Inspect realtime segment size metadata + +For a completed realtime segment, `segment.size.in.bytes` is the logical size of the finalized, uncompressed immutable +segment. Before publishing this value, the committing server applies the latest table index configuration. For example, +if an inverted index was removed while the segment was consuming, the removed index is not included in the published +size. Deep-store recovery also refreshes the value from the immutable segment loaded by the selected server. + +This field is not the compressed archive size and does not represent filesystem-allocated blocks reported by tools +such as `du`. A later segment reload can also make replica sizes differ from the shared metadata snapshot. + +To compare the metadata snapshot with the currently loaded copies of one realtime segment: + +```bash +CONTROLLER='http://localhost:9000' +TABLE='events' +SEGMENT='events__0__0__20260714T0000Z' + +curl -sS "$CONTROLLER/segments/${TABLE}_REALTIME/$SEGMENT/metadata" \ + | jq '{segmentSizeInBytes: (."segment.size.in.bytes" | tonumber)}' + +curl -sS "$CONTROLLER/tables/$TABLE/size?verbose=true" \ + | jq --arg segment "$SEGMENT" \ + '.realtimeSegments.segments[$segment].serverInfo + | to_entries[] + | {server: .key, diskSizeInBytes: .value.diskSizeInBytes}' +``` + +The first value is one shared segment metadata snapshot and does not include replication. The second request reports +each loaded replica separately. + ## Query statistics Request the table-size summary and per-segment details: