Fix realtime segment size metadata

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

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<String, String> 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<String, String> 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);

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,

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);

pinot-core/src/test/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManagerTest.java

@@ -48,13 +48,16 @@
 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;
 import org.apache.pinot.segment.local.segment.index.loader.IndexLoadingConfig;
 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");

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);
   }
 }

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));
     }

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: