[PR] Add FieldConfig.consumingOverride for realtime mutable consuming segments [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 opened a new pull request, #18406:
URL: https://github.com/apache/pinot/pull/18406

   ## Summary
   
   Adds an opt-in per-column override (`FieldConfig.consumingOverride`) that lets a realtime table column have a richer in-memory shape on the **mutable consuming segment** than what is persisted on disk.
   
   **Typical use case:** keep the table in raw encoding (smaller storage, no dictionary on disk) while giving the *consuming* segment a dictionary + inverted index for fast filtering during the consumption window. The committed/immutable segment uses the un-overridden table config — the override never leaks to disk.
   
   ### Example
   
   ```json
   {
     \"fieldConfigList\": [{
       \"name\": \"userId\",
       \"encodingType\": \"RAW\",
       \"consumingOverride\": {
         \"encodingType\": \"DICTIONARY\",
         \"indexTypes\": [\"INVERTED\"]
       }
     }],
     \"tableIndexConfig\": {
       \"noDictionaryColumns\": [\"userId\"]
     }
   }
   ```
   
   ## Architecture
   
   - **SPI**: `FieldConfig.consumingOverride` is an optional `JsonNode` mirroring a partial `FieldConfig` (allowed keys: `encodingType`, `indexTypes`, `indexes`, `compressionCodec`). Backward compatible — older clients see it as an unknown property and ignore it via `@JsonIgnoreProperties(ignoreUnknown=true)` on `BaseJsonConfig`.
   - **Single dispatch point**: `TableConfigUtils.buildConsumingSegmentConfigBuilder` is used by both `RealtimeSegmentDataManager` and `StatelessRealtimeSegmentWriter`.
     - **No override**: returns the existing `IndexLoadingConfig`-driven Builder unchanged (zero behavior change for tables that don't opt in).
     - **With override**: routes the override-applied `TableConfig` back through a new `IndexLoadingConfig` (with a defensive schema clone) so instance config, tier overlay, and schema-derived timestamp index materialization all flow through unchanged.
     - **On `RuntimeException`** (a misconfiguration that slipped past validate): logs and falls back to the persisted shape so consumption keeps making forward progress.
   - **Merge utility**: `TableConfigUtils.applyConsumingOverrides` deep-copies the JSON tree before mutating to avoid aliasing into the cached `TableConfig` (other server threads may be reading it concurrently).
   - **Validation**: shared `enforceConsumingOverrideInvariants` helper used by both validate-time and apply-time paths so the rules can't drift. Rejects non-realtime tables, sorted columns, partition columns, and unknown override keys. Merged shape is re-validated with diagnostics that point back to the originating `consumingOverride`.
   - **Scrub policy**: deny-list — any per-column string-array in `tableIndexConfig` (e.g. `invertedIndexColumns`, `noDictionaryColumns`, ...) is scrubbed for overridden columns. Object-keyed maps and structurally-tied collections (`sortedColumn`, `starTreeIndexConfigs`, `tierOverwrites`) are intentionally left alone to avoid over-scrubbing feature configs whose keys could collide with column names.
   
   ## SPI change note
   
   This adds a new optional field to `FieldConfig` (in `pinot-spi`). Plugin teams that build `FieldConfig` directly via the canonical constructor will see a new 10-arg `@JsonCreator` constructor; the previous 9-arg signature is preserved as a `@Deprecated` overload for source/binary compatibility. Mixed-version clusters: an older server reading a controller-written `consumingOverride` silently drops the unknown field and behaves as if no override exists.
   
   ## Test plan
   
   - [x] 23 unit tests in `TableConfigConsumingOverrideTest` covering:
     - merge semantics (encoding/indexes/compressionCodec replacement, scrub of legacy per-column lists)
     - input-mutation guard (cached `TableConfig` byte-for-byte unchanged after merge)
     - idempotence
     - validation rules (REALTIME-only, sorted-column rejection, partition-column rejection, allowlist enforcement)
     - validate-path *and* apply-path enforcement
     - dispatch helper (no-override, valid override, fallback on failure)
     - JSON round-trip
     - legacy `FieldConfig` JSON without `consumingOverride` still deserializes
   - [x] End-to-end converter test (`RealtimeSegmentConverterTest.testConsumingOverrideKeepsRawShapeOnImmutableSegment`) verifies the immutable segment retains the persisted RAW shape with no inverted index, even when the consuming segment was built with `DICTIONARY + INVERTED`.
   - [x] Spotless / Checkstyle / License all pass.
   - [x] No regression in `TableConfigUtilsTest` (56 tests), `IndexLoadingConfigTest` (3), `BaseTableDataManagerTest` (25), `RealtimeSegmentConverterTest` (161).
   
   🤖 Generated with [Claude Code](https://claude.com/claude-code)


-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add FieldConfig.consumingOverride for richer in-memory shape on realtime consuming segments [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 commented on code in PR #18406:
URL: https://github.com/apache/pinot/pull/18406#discussion_r3201306396


##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -2111,6 +2115,372 @@ private static void overwriteConfig(JsonNode oldCfg, JsonNode newCfg) {
     }
   }
 
+  // Top-level keys in {@code tableIndexConfig} that look like per-column collections (string-array of column
+  // names) but must NOT be touched by the consuming-override scrub. Anything else of string-array shape is treated
+  // as a per-column list and the overridden column name is stripped from it. Inverting the policy this way
+  // (deny-list rather than allow-list) keeps the scrub correct as new per-column index config keys are added to
+  // {@link IndexingConfig} — there is no hand-curated list to drift out of sync. Today every string-array in
+  // {@link IndexingConfig} IS a per-column list (`invertedIndexColumns`, `noDictionaryColumns`,
+  // `bloomFilterColumns`, `rangeIndexColumns`, `jsonIndexColumns`, `onHeapDictionaryColumns`,
+  // `varLengthDictionaryColumns`), so this is correct. **Important:** any future non-per-column string-array field
+  // added to {@link IndexingConfig} MUST be added to this exclusion set, otherwise the scrub will silently strip
+  // entries from it whenever an overridden column name happens to match.
+  //
+  // - sortedColumn is structurally tied to the segment layout; a separate validate-time check rejects overrides
+  //   on sorted columns.
+  // - starTreeIndexConfigs is a list of objects (not column names); leave it alone.
+  // - tierOverwrites is the tier-overlay JSON, handled by a separate transform.
+  private static final Set<String> CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS = Set.of(
+      "sortedColumn",
+      "starTreeIndexConfigs",
+      "tierOverwrites");
+
+  /// Returns a [TableConfig] with each [FieldConfig#getConsumingOverride()] merged into its parent
+  /// [FieldConfig]. The result is intended for building the realtime *mutable consuming* segment only.
+  /// The committed/immutable segment and all loaded immutable segments use the un-overridden table config — that is
+  /// the persisted shape on disk. A typical use is keeping the table in raw encoding for storage efficiency while
+  /// giving the consuming segment a dictionary + inverted index for fast filtering during the consuming window.
+  ///
+  /// Supported override keys: `encodingType`, `indexes`. Other [FieldConfig] fields are intentionally not
+  /// overridable — see [#ALLOWED_CONSUMING_OVERRIDE_KEYS] for the rationale.
+  ///
+  /// Merge semantics: every supported field present in `consumingOverride` replaces the parent field wholesale.
+  /// For `indexes` the entire JSON subtree is replaced rather than merged key-by-key.
+  ///
+  /// For each overridden column, the column is also scrubbed from legacy per-column lists in `tableIndexConfig`
+  /// (e.g. `invertedIndexColumns`, `rangeIndexColumns`, ...) so that those lists do not contradict the merged
+  /// [FieldConfig].
+  ///
+  /// If no [FieldConfig] declares a consuming override, the original [TableConfig] is returned unchanged.
+  ///
+  /// @param tableConfig original table config; not mutated
+  /// @return a new table config with overrides merged, or the original config when no overrides are configured
+  public static TableConfig applyConsumingOverrides(TableConfig tableConfig) {
+    if (!hasConsumingOverride(tableConfig)) {
+      return tableConfig;
+    }
+    // Enforce the same invariants that {@link #validateConsumingOverrides} enforces, so a hand-edited or
+    // older-controller-written TableConfig can't reach the merge with an unsupported override and silently produce
+    // a wrong-shape consuming segment. We re-check here (rather than relying solely on validate-time) because the
+    // merge runs on every server at consuming-segment build time and may see stale configs.
+    enforceConsumingOverrideInvariants(tableConfig);
+    try {
+      // Deep-copy the JSON tree before mutating it. {@code toJsonNode} may alias subtrees on JsonNode-typed fields
+      // of FieldConfig (notably {@code _indexes}, {@code _tierOverwrites}, {@code _consumingOverride}); mutating
+      // those in place would silently corrupt the cached TableConfig that other threads are concurrently reading.
+      JsonNode tblCfgJson = tableConfig.toJsonNode().deepCopy();
+      JsonNode fieldCfgListJson = tblCfgJson.get(TableConfig.FIELD_CONFIG_LIST_KEY);
+      // hasConsumingOverride already proved at least one FieldConfig has a non-empty override, so the JSON view
+      // must surface the field config list as an array. If not, the TableConfig (de)serialization layer is
+      // inconsistent — fail loudly rather than silently shipping the un-overridden shape.
+      Preconditions.checkState(fieldCfgListJson != null && fieldCfgListJson.isArray(),
+          "fieldConfigList missing from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      JsonNode tblIdxCfgJson = tblCfgJson.get(TableConfig.INDEXING_CONFIG_KEY);
+      Set<String> overriddenColumns = new HashSet<>();
+      Iterator<JsonNode> fieldCfgItr = fieldCfgListJson.elements();
+      while (fieldCfgItr.hasNext()) {
+        JsonNode fieldCfgJson = fieldCfgItr.next();
+        JsonNode overrideJson = fieldCfgJson.get(TableConfig.CONSUMING_OVERRIDE_KEY);
+        if (overrideJson == null || !overrideJson.isObject() || overrideJson.isEmpty()) {
+          continue;
+        }
+        JsonNode nameNode = fieldCfgJson.get("name");
+        // FieldConfig's @JsonCreator already guarantees name is non-null, but we re-parsed via toJsonNode so guard
+        // defensively. A missing name is a bug — fail loudly rather than silently dropping the override.
+        Preconditions.checkState(nameNode != null && nameNode.isTextual(),
+            "FieldConfig with consumingOverride is missing a textual 'name' field in table: %s",
+            tableConfig.getTableName());
+        overriddenColumns.add(nameNode.asText());
+        overwriteConfig(fieldCfgJson, overrideJson);
+        // Remove the override marker so downstream consumers see only the merged shape.
+        ((ObjectNode) fieldCfgJson).remove(TableConfig.CONSUMING_OVERRIDE_KEY);
+      }
+      Preconditions.checkState(!overriddenColumns.isEmpty(),
+          "no overridden columns produced from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      if (tblIdxCfgJson != null && tblIdxCfgJson.isObject()) {
+        scrubOverriddenColumnsFromTableIndexConfig((ObjectNode) tblIdxCfgJson, overriddenColumns);
+      }
+      return JsonUtils.jsonNodeToObject(tblCfgJson, TableConfig.class);
+    } catch (IOException e) {
+      // Surface override parse/deserialization failures (e.g. unknown keys, type mismatches) loudly so the bad
+      // config is rejected at validate time rather than silently falling back to the un-overridden shape — that
+      // would build the consuming segment with the wrong layout and the user would get no signal.
+      throw new IllegalStateException(
+          "Failed to apply consuming overrides for table: " + tableConfig.getTableName(), e);
+    }
+  }
+
+  /// For each top-level entry in `tableIndexConfig` that holds an array of column-name strings (legacy
+  /// per-column lists like `invertedIndexColumns`, `noDictionaryColumns`, etc.), strip any name in
+  /// `overriddenColumns`. Entries whose key is in [#CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS], that are not arrays,
+  /// or that contain non-string elements, are left alone.
+  ///
+  /// Object-shaped values (e.g. `segmentPartitionConfig.columnPartitionMap`, feature configs) are intentionally
+  /// NOT scrubbed: the heuristic "drop any object key matching an overridden column name" would silently drop
+  /// unrelated feature-config keys whose names happen to collide with a column name. Validation already rejects
+  /// overrides on partition columns up front.
+  private static void scrubOverriddenColumnsFromTableIndexConfig(ObjectNode tblIdxCfgObj,
+      Set<String> overriddenColumns) {
+    Iterator<Map.Entry<String, JsonNode>> entries = tblIdxCfgObj.properties().iterator();
+    while (entries.hasNext()) {
+      Map.Entry<String, JsonNode> entry = entries.next();
+      String key = entry.getKey();
+      if (CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS.contains(key)) {
+        continue;
+      }
+      JsonNode value = entry.getValue();
+      if (value == null || !value.isArray() || value.isEmpty()) {

Review Comment:
   This scrub only handles array-valued `tableIndexConfig` entries, but `ForwardIndexType` still treats `noDictionaryConfig.keySet()` as the RAW signal. That means a table using `noDictionaryConfig` for the persisted RAW column cannot opt into `consumingOverride.encodingType=DICTIONARY`: the merged config still carries the map entry and `validateIndexingConfigAndFieldConfigList` will reject it as a DICTIONARY-vs-noDictionary conflict. Please scrub `noDictionaryConfig` for overridden columns too, otherwise the documented RAW-on-disk / DICTIONARY-on-consuming use case only works for `noDictionaryColumns`.



##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -2111,6 +2115,372 @@ private static void overwriteConfig(JsonNode oldCfg, JsonNode newCfg) {
     }
   }
 
+  // Top-level keys in {@code tableIndexConfig} that look like per-column collections (string-array of column
+  // names) but must NOT be touched by the consuming-override scrub. Anything else of string-array shape is treated
+  // as a per-column list and the overridden column name is stripped from it. Inverting the policy this way
+  // (deny-list rather than allow-list) keeps the scrub correct as new per-column index config keys are added to
+  // {@link IndexingConfig} — there is no hand-curated list to drift out of sync. Today every string-array in
+  // {@link IndexingConfig} IS a per-column list (`invertedIndexColumns`, `noDictionaryColumns`,
+  // `bloomFilterColumns`, `rangeIndexColumns`, `jsonIndexColumns`, `onHeapDictionaryColumns`,
+  // `varLengthDictionaryColumns`), so this is correct. **Important:** any future non-per-column string-array field
+  // added to {@link IndexingConfig} MUST be added to this exclusion set, otherwise the scrub will silently strip
+  // entries from it whenever an overridden column name happens to match.
+  //
+  // - sortedColumn is structurally tied to the segment layout; a separate validate-time check rejects overrides
+  //   on sorted columns.
+  // - starTreeIndexConfigs is a list of objects (not column names); leave it alone.
+  // - tierOverwrites is the tier-overlay JSON, handled by a separate transform.
+  private static final Set<String> CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS = Set.of(
+      "sortedColumn",
+      "starTreeIndexConfigs",
+      "tierOverwrites");
+
+  /// Returns a [TableConfig] with each [FieldConfig#getConsumingOverride()] merged into its parent
+  /// [FieldConfig]. The result is intended for building the realtime *mutable consuming* segment only.
+  /// The committed/immutable segment and all loaded immutable segments use the un-overridden table config — that is
+  /// the persisted shape on disk. A typical use is keeping the table in raw encoding for storage efficiency while
+  /// giving the consuming segment a dictionary + inverted index for fast filtering during the consuming window.
+  ///
+  /// Supported override keys: `encodingType`, `indexes`. Other [FieldConfig] fields are intentionally not
+  /// overridable — see [#ALLOWED_CONSUMING_OVERRIDE_KEYS] for the rationale.
+  ///
+  /// Merge semantics: every supported field present in `consumingOverride` replaces the parent field wholesale.
+  /// For `indexes` the entire JSON subtree is replaced rather than merged key-by-key.
+  ///
+  /// For each overridden column, the column is also scrubbed from legacy per-column lists in `tableIndexConfig`
+  /// (e.g. `invertedIndexColumns`, `rangeIndexColumns`, ...) so that those lists do not contradict the merged
+  /// [FieldConfig].
+  ///
+  /// If no [FieldConfig] declares a consuming override, the original [TableConfig] is returned unchanged.
+  ///
+  /// @param tableConfig original table config; not mutated
+  /// @return a new table config with overrides merged, or the original config when no overrides are configured
+  public static TableConfig applyConsumingOverrides(TableConfig tableConfig) {
+    if (!hasConsumingOverride(tableConfig)) {
+      return tableConfig;
+    }
+    // Enforce the same invariants that {@link #validateConsumingOverrides} enforces, so a hand-edited or
+    // older-controller-written TableConfig can't reach the merge with an unsupported override and silently produce
+    // a wrong-shape consuming segment. We re-check here (rather than relying solely on validate-time) because the
+    // merge runs on every server at consuming-segment build time and may see stale configs.
+    enforceConsumingOverrideInvariants(tableConfig);
+    try {
+      // Deep-copy the JSON tree before mutating it. {@code toJsonNode} may alias subtrees on JsonNode-typed fields
+      // of FieldConfig (notably {@code _indexes}, {@code _tierOverwrites}, {@code _consumingOverride}); mutating
+      // those in place would silently corrupt the cached TableConfig that other threads are concurrently reading.
+      JsonNode tblCfgJson = tableConfig.toJsonNode().deepCopy();
+      JsonNode fieldCfgListJson = tblCfgJson.get(TableConfig.FIELD_CONFIG_LIST_KEY);
+      // hasConsumingOverride already proved at least one FieldConfig has a non-empty override, so the JSON view
+      // must surface the field config list as an array. If not, the TableConfig (de)serialization layer is
+      // inconsistent — fail loudly rather than silently shipping the un-overridden shape.
+      Preconditions.checkState(fieldCfgListJson != null && fieldCfgListJson.isArray(),
+          "fieldConfigList missing from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      JsonNode tblIdxCfgJson = tblCfgJson.get(TableConfig.INDEXING_CONFIG_KEY);
+      Set<String> overriddenColumns = new HashSet<>();
+      Iterator<JsonNode> fieldCfgItr = fieldCfgListJson.elements();
+      while (fieldCfgItr.hasNext()) {
+        JsonNode fieldCfgJson = fieldCfgItr.next();
+        JsonNode overrideJson = fieldCfgJson.get(TableConfig.CONSUMING_OVERRIDE_KEY);
+        if (overrideJson == null || !overrideJson.isObject() || overrideJson.isEmpty()) {
+          continue;
+        }
+        JsonNode nameNode = fieldCfgJson.get("name");
+        // FieldConfig's @JsonCreator already guarantees name is non-null, but we re-parsed via toJsonNode so guard
+        // defensively. A missing name is a bug — fail loudly rather than silently dropping the override.
+        Preconditions.checkState(nameNode != null && nameNode.isTextual(),
+            "FieldConfig with consumingOverride is missing a textual 'name' field in table: %s",
+            tableConfig.getTableName());
+        overriddenColumns.add(nameNode.asText());
+        overwriteConfig(fieldCfgJson, overrideJson);
+        // Remove the override marker so downstream consumers see only the merged shape.
+        ((ObjectNode) fieldCfgJson).remove(TableConfig.CONSUMING_OVERRIDE_KEY);
+      }
+      Preconditions.checkState(!overriddenColumns.isEmpty(),
+          "no overridden columns produced from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      if (tblIdxCfgJson != null && tblIdxCfgJson.isObject()) {
+        scrubOverriddenColumnsFromTableIndexConfig((ObjectNode) tblIdxCfgJson, overriddenColumns);
+      }
+      return JsonUtils.jsonNodeToObject(tblCfgJson, TableConfig.class);
+    } catch (IOException e) {
+      // Surface override parse/deserialization failures (e.g. unknown keys, type mismatches) loudly so the bad
+      // config is rejected at validate time rather than silently falling back to the un-overridden shape — that
+      // would build the consuming segment with the wrong layout and the user would get no signal.
+      throw new IllegalStateException(
+          "Failed to apply consuming overrides for table: " + tableConfig.getTableName(), e);
+    }
+  }
+
+  /// For each top-level entry in `tableIndexConfig` that holds an array of column-name strings (legacy
+  /// per-column lists like `invertedIndexColumns`, `noDictionaryColumns`, etc.), strip any name in
+  /// `overriddenColumns`. Entries whose key is in [#CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS], that are not arrays,
+  /// or that contain non-string elements, are left alone.
+  ///
+  /// Object-shaped values (e.g. `segmentPartitionConfig.columnPartitionMap`, feature configs) are intentionally
+  /// NOT scrubbed: the heuristic "drop any object key matching an overridden column name" would silently drop
+  /// unrelated feature-config keys whose names happen to collide with a column name. Validation already rejects
+  /// overrides on partition columns up front.
+  private static void scrubOverriddenColumnsFromTableIndexConfig(ObjectNode tblIdxCfgObj,
+      Set<String> overriddenColumns) {
+    Iterator<Map.Entry<String, JsonNode>> entries = tblIdxCfgObj.properties().iterator();
+    while (entries.hasNext()) {
+      Map.Entry<String, JsonNode> entry = entries.next();
+      String key = entry.getKey();
+      if (CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS.contains(key)) {
+        continue;
+      }
+      JsonNode value = entry.getValue();
+      if (value == null || !value.isArray() || value.isEmpty()) {
+        continue;
+      }
+      ArrayNode arr = (ArrayNode) value;
+      for (JsonNode element : arr) {
+        if (!element.isTextual()) {
+          // Non-string-array entry — not a per-column list; leave alone.
+          arr = null;
+          break;
+        }
+      }
+      if (arr == null) {
+        continue;
+      }
+      List<JsonNode> kept = new ArrayList<>(arr.size());
+      for (JsonNode element : arr) {
+        if (!overriddenColumns.contains(element.asText())) {
+          kept.add(element);
+        }
+      }
+      if (kept.size() != arr.size()) {
+        ArrayNode replacement = JsonUtils.newArrayNode();
+        for (JsonNode element : kept) {
+          replacement.add(element);
+        }
+        tblIdxCfgObj.set(key, replacement);
+      }
+    }
+  }
+
+  /// Builds the [RealtimeSegmentConfig.Builder] used to construct a mutable consuming segment, applying any
+  /// [FieldConfig#getConsumingOverride()] present on the table config. When no override is configured, the
+  /// un-modified [IndexLoadingConfig]-driven builder is returned so existing code paths (tier overwrites,
+  /// instance-level mutations on `IndexLoadingConfig`) flow through unchanged. If the override merge throws (a
+  /// misconfiguration that slipped past validate), the call falls back to the `IndexLoadingConfig`-driven builder,
+  /// logs the error, and invokes `onFallback` so the caller can emit a metric / surface the degradation. The
+  /// fallback-applied builder produces a consuming segment matching the persisted shape (no override) for the
+  /// full lifetime of that consuming segment — there is no auto-retry; operators must reload the table after
+  /// fixing the override config to pick up the corrected shape on the next consuming-segment build.
+  ///
+  /// **IndexLoadingConfig contract on the override path:** the helper rebuilds a fresh [IndexLoadingConfig] from
+  /// `(indexLoadingConfig.getInstanceDataManagerConfig(), mergedTableConfig, schemaCopy)` and re-applies the
+  /// `segmentTier`. Any other in-place mutation the caller made on the supplied `indexLoadingConfig` (segment
+  /// version, star-tree configs, etc.) is **not** carried over on the override path. Callers that need such
+  /// mutations preserved must either set them on the returned Builder after this call, or extend this helper.
+  ///
+  /// **Schema clone:** the override path defensively deep-copies the schema via JSON round-trip because the
+  /// override-path `IndexLoadingConfig` constructor invokes `TimestampIndexUtils.applyTimestampIndex` which
+  /// mutates the schema under a lock keyed to the *new* (per-call) TableConfig instance. The non-override path
+  /// constructs `IndexLoadingConfig` once at table-data-manager-load time (not per-segment-build), so the lock
+  /// is not racing per call there. The clone is therefore needed only on the override path. TODO: fix
+  /// TimestampIndexUtils to not mutate its inputs so this clone can be removed.
+  ///
+  /// @param tableConfig source table config; not mutated
+  /// @param schema      schema in scope; deep-copied before being passed to the override-path IndexLoadingConfig
+  /// @param indexLoadingConfig fallback / no-override path; on the override path only `instanceDataManagerConfig`
+  ///                           and `segmentTier` are read (see contract above)
+  /// @param logger      logger to emit error on fallback
+  /// @param onFallback  optional callback invoked when the override merge fails and the helper falls back to the
+  ///                    persisted shape; intended for the caller to bump a metric (e.g.
+  ///                    [ServerMeter#CONSUMING_OVERRIDE_FALLBACK]) tagged with its own table-name context. May
+  ///                    be {@code null} to skip metric emission (e.g. in unit tests).
+  /// @return a Builder ready for the caller to chain `.set...` calls and `.build()`
+  public static RealtimeSegmentConfig.Builder buildConsumingSegmentConfigBuilder(TableConfig tableConfig,
+      Schema schema, IndexLoadingConfig indexLoadingConfig, Logger logger, @Nullable Runnable onFallback) {
+    if (hasConsumingOverride(tableConfig)) {
+      try {
+        TableConfig consumingTableConfig = applyConsumingOverrides(tableConfig);
+        Schema schemaCopy;
+        try {
+          schemaCopy = JsonUtils.jsonNodeToObject(schema.toJsonObject(), Schema.class);
+        } catch (IOException ioe) {
+          throw new IllegalStateException("Failed to clone schema for consumingOverride path on table: "
+              + tableConfig.getTableName(), ioe);
+        }
+        IndexLoadingConfig consumingIlc = new IndexLoadingConfig(indexLoadingConfig.getInstanceDataManagerConfig(),
+            consumingTableConfig, schemaCopy);
+        /// `segmentVersion` and `readMode` are already derived by the constructor from the same
+        /// `instanceDataManagerConfig` + `consumingTableConfig.indexingConfig` inputs the source ILC used, so they
+        /// don't need to be copied across. Tier overlay, however, is set by the caller post-construction (it is
+        /// not in either input), so we re-apply it explicitly.
+        String segmentTier = indexLoadingConfig.getSegmentTier();
+        if (segmentTier != null) {
+          consumingIlc.setSegmentTier(segmentTier);
+        }
+        return new RealtimeSegmentConfig.Builder(consumingIlc);
+      } catch (RuntimeException e) {
+        /// Include the override snippet so the consuming-segment build log line is self-sufficient for triage —
+        /// operators reading the log do not need to also fetch the table config from ZK to see which override
+        /// failed. The metric (CONSUMING_OVERRIDE_FALLBACK) is the alerting hook; this log is the diagnostic.
+        logger.error("Failed to apply consumingOverride for table: {} (override snippet: {}); falling back to "
+            + "persisted shape — consuming segment will run with the un-overridden shape for its full lifetime",
+            tableConfig.getTableName(), summarizeConsumingOverrides(tableConfig), e);
+        if (onFallback != null) {
+          onFallback.run();
+        }
+      }
+    }
+    return new RealtimeSegmentConfig.Builder(indexLoadingConfig);
+  }
+
+  /// Returns a `column → override` short summary for the failure log.
+  private static String summarizeConsumingOverrides(TableConfig tableConfig) {
+    List<FieldConfig> fieldConfigs = tableConfig.getFieldConfigList();
+    if (fieldConfigs == null) {
+      return "{}";
+    }
+    StringBuilder sb = new StringBuilder("{");
+    boolean first = true;
+    for (FieldConfig fieldConfig : fieldConfigs) {
+      JsonNode override = fieldConfig.getConsumingOverride();
+      if (override == null || !override.isObject() || override.isEmpty()) {
+        continue;
+      }
+      if (!first) {
+        sb.append(", ");
+      }
+      sb.append(fieldConfig.getName()).append('=').append(override);
+      first = false;
+    }
+    sb.append('}');
+    return sb.toString();
+  }
+
+  public static boolean hasConsumingOverride(TableConfig tableConfig) {
+    List<FieldConfig> fieldConfigList = tableConfig.getFieldConfigList();
+    if (fieldConfigList == null) {
+      return false;
+    }
+    for (FieldConfig fieldConfig : fieldConfigList) {
+      JsonNode overrideJson = fieldConfig.getConsumingOverride();
+      if (overrideJson != null && overrideJson.isObject() && !overrideJson.isEmpty()) {

Review Comment:
   Non-object `consumingOverride` values are silently treated as "no override" here. For example, `consumingOverride: []` or `consumingOverride: "foo"` passes validation, then the realtime path takes the default builder with no error, log, or metric. Since this is a user-facing config and the rest of the patch is trying to fail fast on bad override shapes, please reject any non-null `consumingOverride` that is not an object instead of letting `hasConsumingOverride()` filter it out.



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add FieldConfig.consumingOverride for realtime mutable consuming segments [pinot]

["codecov-commenter (via GitHub)" <gi...@apache.org>]

codecov-commenter commented on PR #18406:
URL: https://github.com/apache/pinot/pull/18406#issuecomment-4366271843

   ## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/18406?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
   :x: Patch coverage is `15.33333%` with `127 lines` in your changes missing coverage. Please review.
   :white_check_mark: Project coverage is 55.40%. Comparing base ([`80fb0e8`](https://app.codecov.io/gh/apache/pinot/commit/80fb0e8932aa9b1beb40df9613111667b8b5765d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`31ea683`](https://app.codecov.io/gh/apache/pinot/commit/31ea6835909765e9e875190c3c23fd206b5d810f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
   :warning: Report is 1 commits behind head on master.
   
   | [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/18406?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
   |---|---|---|
   | [...he/pinot/segment/local/utils/TableConfigUtils.java](https://app.codecov.io/gh/apache/pinot/pull/18406?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Futils%2FTableConfigUtils.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC91dGlscy9UYWJsZUNvbmZpZ1V0aWxzLmphdmE=) | 9.02% | [118 Missing and 3 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18406?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   | [...ealtime/writer/StatelessRealtimeSegmentWriter.java](https://app.codecov.io/gh/apache/pinot/pull/18406?src=pr&el=tree&filepath=pinot-segment-local%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fsegment%2Flocal%2Frealtime%2Fwriter%2FStatelessRealtimeSegmentWriter.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc2VnbWVudC1sb2NhbC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3Qvc2VnbWVudC9sb2NhbC9yZWFsdGltZS93cml0ZXIvU3RhdGVsZXNzUmVhbHRpbWVTZWdtZW50V3JpdGVyLmphdmE=) | 0.00% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18406?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   | [...org/apache/pinot/spi/config/table/FieldConfig.java](https://app.codecov.io/gh/apache/pinot/pull/18406?src=pr&el=tree&filepath=pinot-spi%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fspi%2Fconfig%2Ftable%2FFieldConfig.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3Qtc3BpL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9waW5vdC9zcGkvY29uZmlnL3RhYmxlL0ZpZWxkQ29uZmlnLmphdmE=) | 72.72% | [3 Missing :warning: ](https://app.codecov.io/gh/apache/pinot/pull/18406?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
   > :exclamation:  There is a different number of reports uploaded between BASE (80fb0e8) and HEAD (31ea683). Click for more details.
   > 
   > <details><summary>HEAD has 16 uploads less than BASE</summary>
   >
   >| Flag | BASE (80fb0e8) | HEAD (31ea683) |
   >|------|------|------|
   >|java-21|5|1|
   >|unittests|2|1|
   >|temurin|5|1|
   >|unittests2|1|0|
   >|integration|3|0|
   >|integration1|1|0|
   >|integration2|1|0|
   >|custom-integration1|1|0|
   ></details>
   
   <details><summary>Additional details and impacted files</summary>
   
   
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #18406      +/-   ##
   ============================================
   - Coverage     63.46%   55.40%   -8.06%     
   + Complexity     1701      860     -841     
   ============================================
     Files          3254     2544     -710     
     Lines        199104   147830   -51274     
     Branches      30830    23827    -7003     
   ============================================
   - Hits         126354    81900   -44454     
   + Misses        62665    58908    -3757     
   + Partials      10085     7022    -3063     
   ```
   
   | [Flag](https://app.codecov.io/gh/apache/pinot/pull/18406/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
   |---|---|---|
   | [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/18406/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration](https://app.codecov.io/gh/apache/pinot/pull/18406/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration1](https://app.codecov.io/gh/apache/pinot/pull/18406/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [integration2](https://app.codecov.io/gh/apache/pinot/pull/18406/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   | [java-21](https://app.codecov.io/gh/apache/pinot/pull/18406/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.40% <15.33%> (-8.06%)` | :arrow_down: |
   | [temurin](https://app.codecov.io/gh/apache/pinot/pull/18406/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.40% <15.33%> (-8.06%)` | :arrow_down: |
   | [unittests](https://app.codecov.io/gh/apache/pinot/pull/18406/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.40% <15.33%> (-8.06%)` | :arrow_down: |
   | [unittests1](https://app.codecov.io/gh/apache/pinot/pull/18406/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `55.40% <15.33%> (-0.02%)` | :arrow_down: |
   | [unittests2](https://app.codecov.io/gh/apache/pinot/pull/18406/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
   </details>
   
   [:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/18406?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).   
   :loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
   <details><summary> :rocket: New features to boost your workflow: </summary>
   
   - :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
   </details>


-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add FieldConfig.consumingOverride for richer in-memory shape on realtime consuming segments [pinot]

["xiangfu0 (via GitHub)" <gi...@apache.org>]

xiangfu0 closed pull request #18406: Add FieldConfig.consumingOverride for richer in-memory shape on realtime consuming segments
URL: https://github.com/apache/pinot/pull/18406


-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org

Re: [PR] Add FieldConfig.consumingOverride for richer in-memory shape on realtime consuming segments [pinot]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #18406:
URL: https://github.com/apache/pinot/pull/18406#discussion_r3180004413


##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -2111,6 +2115,318 @@ private static void overwriteConfig(JsonNode oldCfg, JsonNode newCfg) {
     }
   }
 
+  // Top-level keys in {@code tableIndexConfig} that look like per-column collections (string-array or string-keyed
+  // object) but must NOT be touched by the consuming-override scrub. Anything else of those shapes is treated as
+  // a per-column collection and the overridden column name is stripped from it. Inverting the policy this way
+  // (deny-list rather than allow-list) keeps the scrub correct as new per-column index config keys are added to
+  // {@link IndexingConfig} — there is no hand-curated list to drift out of sync.
+  //
+  // - sortedColumn is structurally tied to the segment layout; a separate validate-time check rejects overrides
+  //   on sorted columns.
+  // - starTreeIndexConfigs is a list of objects (not column names); leave it alone.
+  // - tierOverwrites is the tier-overlay JSON, handled by a separate transform.
+  private static final Set<String> CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS = Set.of(
+      "sortedColumn",
+      "starTreeIndexConfigs",
+      "tierOverwrites");
+
+  /// Returns a [TableConfig] with each [FieldConfig#getConsumingOverride()] merged into its parent
+  /// [FieldConfig]. The result is intended for building the realtime *mutable consuming* segment only.
+  /// The committed/immutable segment and all loaded immutable segments use the un-overridden table config — that is
+  /// the persisted shape on disk. A typical use is keeping the table in raw encoding for storage efficiency while
+  /// giving the consuming segment a dictionary + inverted index for fast filtering during the consuming window.
+  ///
+  /// Supported override keys: `encodingType`, `indexes`. Other [FieldConfig] fields are intentionally not
+  /// overridable — see [#ALLOWED_CONSUMING_OVERRIDE_KEYS] for the rationale.
+  ///
+  /// Merge semantics: every supported field present in `consumingOverride` replaces the parent field wholesale.
+  /// For `indexes` the entire JSON subtree is replaced rather than merged key-by-key.
+  ///
+  /// For each overridden column, the column is also scrubbed from legacy per-column lists in `tableIndexConfig`
+  /// (e.g. `invertedIndexColumns`, `rangeIndexColumns`, ...) so that those lists do not contradict the merged
+  /// [FieldConfig].
+  ///
+  /// If no [FieldConfig] declares a consuming override, the original [TableConfig] is returned unchanged.
+  ///
+  /// @param tableConfig original table config; not mutated
+  /// @return a new table config with overrides merged, or the original config when no overrides are configured
+  public static TableConfig applyConsumingOverrides(TableConfig tableConfig) {
+    if (!hasConsumingOverride(tableConfig)) {
+      return tableConfig;
+    }
+    // Enforce the same invariants that {@link #validateConsumingOverrides} enforces, so a hand-edited or
+    // older-controller-written TableConfig can't reach the merge with an unsupported override and silently produce
+    // a wrong-shape consuming segment. We re-check here (rather than relying solely on validate-time) because the
+    // merge runs on every server at consuming-segment build time and may see stale configs.
+    enforceConsumingOverrideInvariants(tableConfig);
+    try {
+      // Deep-copy the JSON tree before mutating it. {@code toJsonNode} may alias subtrees on JsonNode-typed fields
+      // of FieldConfig (notably {@code _indexes}, {@code _tierOverwrites}, {@code _consumingOverride}); mutating
+      // those in place would silently corrupt the cached TableConfig that other threads are concurrently reading.
+      JsonNode tblCfgJson = tableConfig.toJsonNode().deepCopy();
+      JsonNode fieldCfgListJson = tblCfgJson.get(TableConfig.FIELD_CONFIG_LIST_KEY);
+      // hasConsumingOverride already proved at least one FieldConfig has a non-empty override, so the JSON view
+      // must surface the field config list as an array. If not, the TableConfig (de)serialization layer is
+      // inconsistent — fail loudly rather than silently shipping the un-overridden shape.
+      Preconditions.checkState(fieldCfgListJson != null && fieldCfgListJson.isArray(),
+          "fieldConfigList missing from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      JsonNode tblIdxCfgJson = tblCfgJson.get(TableConfig.INDEXING_CONFIG_KEY);
+      Set<String> overriddenColumns = new HashSet<>();
+      Iterator<JsonNode> fieldCfgItr = fieldCfgListJson.elements();
+      while (fieldCfgItr.hasNext()) {
+        JsonNode fieldCfgJson = fieldCfgItr.next();
+        JsonNode overrideJson = fieldCfgJson.get(TableConfig.CONSUMING_OVERRIDE_KEY);
+        if (overrideJson == null || !overrideJson.isObject() || overrideJson.isEmpty()) {
+          continue;
+        }
+        JsonNode nameNode = fieldCfgJson.get("name");
+        // FieldConfig's @JsonCreator already guarantees name is non-null, but we re-parsed via toJsonNode so guard
+        // defensively. A missing name is a bug — fail loudly rather than silently dropping the override.
+        Preconditions.checkState(nameNode != null && nameNode.isTextual(),
+            "FieldConfig with consumingOverride is missing a textual 'name' field in table: %s",
+            tableConfig.getTableName());
+        overriddenColumns.add(nameNode.asText());

Review Comment:
   Overwriting only the `indexes` JSON leaves any legacy `FieldConfig.indexTypes` values on the parent config untouched. Columns that still use the deprecated `indexTypes` form therefore cannot use `consumingOverride.indexes` as a replacement: the merged config ends up declaring the same index through both the new and legacy paths, which `FieldIndexConfigsUtil` treats as a duplicate definition.
   



##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -2111,6 +2115,318 @@ private static void overwriteConfig(JsonNode oldCfg, JsonNode newCfg) {
     }
   }
 
+  // Top-level keys in {@code tableIndexConfig} that look like per-column collections (string-array or string-keyed
+  // object) but must NOT be touched by the consuming-override scrub. Anything else of those shapes is treated as
+  // a per-column collection and the overridden column name is stripped from it. Inverting the policy this way
+  // (deny-list rather than allow-list) keeps the scrub correct as new per-column index config keys are added to
+  // {@link IndexingConfig} — there is no hand-curated list to drift out of sync.
+  //
+  // - sortedColumn is structurally tied to the segment layout; a separate validate-time check rejects overrides
+  //   on sorted columns.
+  // - starTreeIndexConfigs is a list of objects (not column names); leave it alone.
+  // - tierOverwrites is the tier-overlay JSON, handled by a separate transform.
+  private static final Set<String> CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS = Set.of(
+      "sortedColumn",
+      "starTreeIndexConfigs",
+      "tierOverwrites");
+
+  /// Returns a [TableConfig] with each [FieldConfig#getConsumingOverride()] merged into its parent
+  /// [FieldConfig]. The result is intended for building the realtime *mutable consuming* segment only.
+  /// The committed/immutable segment and all loaded immutable segments use the un-overridden table config — that is
+  /// the persisted shape on disk. A typical use is keeping the table in raw encoding for storage efficiency while
+  /// giving the consuming segment a dictionary + inverted index for fast filtering during the consuming window.
+  ///
+  /// Supported override keys: `encodingType`, `indexes`. Other [FieldConfig] fields are intentionally not
+  /// overridable — see [#ALLOWED_CONSUMING_OVERRIDE_KEYS] for the rationale.
+  ///
+  /// Merge semantics: every supported field present in `consumingOverride` replaces the parent field wholesale.
+  /// For `indexes` the entire JSON subtree is replaced rather than merged key-by-key.
+  ///
+  /// For each overridden column, the column is also scrubbed from legacy per-column lists in `tableIndexConfig`
+  /// (e.g. `invertedIndexColumns`, `rangeIndexColumns`, ...) so that those lists do not contradict the merged
+  /// [FieldConfig].
+  ///
+  /// If no [FieldConfig] declares a consuming override, the original [TableConfig] is returned unchanged.
+  ///
+  /// @param tableConfig original table config; not mutated
+  /// @return a new table config with overrides merged, or the original config when no overrides are configured
+  public static TableConfig applyConsumingOverrides(TableConfig tableConfig) {
+    if (!hasConsumingOverride(tableConfig)) {
+      return tableConfig;
+    }
+    // Enforce the same invariants that {@link #validateConsumingOverrides} enforces, so a hand-edited or
+    // older-controller-written TableConfig can't reach the merge with an unsupported override and silently produce
+    // a wrong-shape consuming segment. We re-check here (rather than relying solely on validate-time) because the
+    // merge runs on every server at consuming-segment build time and may see stale configs.
+    enforceConsumingOverrideInvariants(tableConfig);
+    try {
+      // Deep-copy the JSON tree before mutating it. {@code toJsonNode} may alias subtrees on JsonNode-typed fields
+      // of FieldConfig (notably {@code _indexes}, {@code _tierOverwrites}, {@code _consumingOverride}); mutating
+      // those in place would silently corrupt the cached TableConfig that other threads are concurrently reading.
+      JsonNode tblCfgJson = tableConfig.toJsonNode().deepCopy();
+      JsonNode fieldCfgListJson = tblCfgJson.get(TableConfig.FIELD_CONFIG_LIST_KEY);
+      // hasConsumingOverride already proved at least one FieldConfig has a non-empty override, so the JSON view
+      // must surface the field config list as an array. If not, the TableConfig (de)serialization layer is
+      // inconsistent — fail loudly rather than silently shipping the un-overridden shape.
+      Preconditions.checkState(fieldCfgListJson != null && fieldCfgListJson.isArray(),
+          "fieldConfigList missing from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      JsonNode tblIdxCfgJson = tblCfgJson.get(TableConfig.INDEXING_CONFIG_KEY);
+      Set<String> overriddenColumns = new HashSet<>();
+      Iterator<JsonNode> fieldCfgItr = fieldCfgListJson.elements();
+      while (fieldCfgItr.hasNext()) {
+        JsonNode fieldCfgJson = fieldCfgItr.next();
+        JsonNode overrideJson = fieldCfgJson.get(TableConfig.CONSUMING_OVERRIDE_KEY);
+        if (overrideJson == null || !overrideJson.isObject() || overrideJson.isEmpty()) {
+          continue;
+        }
+        JsonNode nameNode = fieldCfgJson.get("name");
+        // FieldConfig's @JsonCreator already guarantees name is non-null, but we re-parsed via toJsonNode so guard
+        // defensively. A missing name is a bug — fail loudly rather than silently dropping the override.
+        Preconditions.checkState(nameNode != null && nameNode.isTextual(),
+            "FieldConfig with consumingOverride is missing a textual 'name' field in table: %s",
+            tableConfig.getTableName());
+        overriddenColumns.add(nameNode.asText());
+        overwriteConfig(fieldCfgJson, overrideJson);
+        // Remove the override marker so downstream consumers see only the merged shape.
+        ((ObjectNode) fieldCfgJson).remove(TableConfig.CONSUMING_OVERRIDE_KEY);
+      }
+      Preconditions.checkState(!overriddenColumns.isEmpty(),
+          "no overridden columns produced from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      if (tblIdxCfgJson != null && tblIdxCfgJson.isObject()) {
+        scrubOverriddenColumnsFromTableIndexConfig((ObjectNode) tblIdxCfgJson, overriddenColumns);
+      }
+      return JsonUtils.jsonNodeToObject(tblCfgJson, TableConfig.class);
+    } catch (IOException e) {
+      // Surface override parse/deserialization failures (e.g. unknown keys, type mismatches) loudly so the bad
+      // config is rejected at validate time rather than silently falling back to the un-overridden shape — that
+      // would build the consuming segment with the wrong layout and the user would get no signal.
+      throw new IllegalStateException(
+          "Failed to apply consuming overrides for table: " + tableConfig.getTableName(), e);
+    }
+  }
+
+  /// For each top-level entry in `tableIndexConfig` that holds an array of column-name strings (legacy
+  /// per-column lists like `invertedIndexColumns`, `noDictionaryColumns`, etc.), strip any name in
+  /// `overriddenColumns`. Entries whose key is in [#CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS], that are not arrays,
+  /// or that contain non-string elements, are left alone.
+  ///
+  /// Object-shaped values (e.g. `segmentPartitionConfig.columnPartitionMap`, feature configs) are intentionally
+  /// NOT scrubbed: the heuristic "drop any object key matching an overridden column name" would silently drop
+  /// unrelated feature-config keys whose names happen to collide with a column name. Validation already rejects
+  /// overrides on partition columns up front.

Review Comment:
   Skipping all object-shaped `tableIndexConfig` entries breaks currently supported column-keyed configs such as `noDictionaryConfig`. If a table keeps a column RAW via `noDictionaryConfig` and adds `consumingOverride.encodingType=DICTIONARY`, the merged consuming view still carries the legacy no-dictionary entry, so validation/fallback will reject the override even though the persisted shape is valid.



##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -2111,6 +2115,318 @@ private static void overwriteConfig(JsonNode oldCfg, JsonNode newCfg) {
     }
   }
 
+  // Top-level keys in {@code tableIndexConfig} that look like per-column collections (string-array or string-keyed
+  // object) but must NOT be touched by the consuming-override scrub. Anything else of those shapes is treated as
+  // a per-column collection and the overridden column name is stripped from it. Inverting the policy this way
+  // (deny-list rather than allow-list) keeps the scrub correct as new per-column index config keys are added to
+  // {@link IndexingConfig} — there is no hand-curated list to drift out of sync.
+  //
+  // - sortedColumn is structurally tied to the segment layout; a separate validate-time check rejects overrides
+  //   on sorted columns.
+  // - starTreeIndexConfigs is a list of objects (not column names); leave it alone.
+  // - tierOverwrites is the tier-overlay JSON, handled by a separate transform.
+  private static final Set<String> CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS = Set.of(
+      "sortedColumn",
+      "starTreeIndexConfigs",
+      "tierOverwrites");
+
+  /// Returns a [TableConfig] with each [FieldConfig#getConsumingOverride()] merged into its parent
+  /// [FieldConfig]. The result is intended for building the realtime *mutable consuming* segment only.
+  /// The committed/immutable segment and all loaded immutable segments use the un-overridden table config — that is
+  /// the persisted shape on disk. A typical use is keeping the table in raw encoding for storage efficiency while
+  /// giving the consuming segment a dictionary + inverted index for fast filtering during the consuming window.
+  ///
+  /// Supported override keys: `encodingType`, `indexes`. Other [FieldConfig] fields are intentionally not
+  /// overridable — see [#ALLOWED_CONSUMING_OVERRIDE_KEYS] for the rationale.
+  ///
+  /// Merge semantics: every supported field present in `consumingOverride` replaces the parent field wholesale.
+  /// For `indexes` the entire JSON subtree is replaced rather than merged key-by-key.
+  ///
+  /// For each overridden column, the column is also scrubbed from legacy per-column lists in `tableIndexConfig`
+  /// (e.g. `invertedIndexColumns`, `rangeIndexColumns`, ...) so that those lists do not contradict the merged
+  /// [FieldConfig].
+  ///
+  /// If no [FieldConfig] declares a consuming override, the original [TableConfig] is returned unchanged.
+  ///
+  /// @param tableConfig original table config; not mutated
+  /// @return a new table config with overrides merged, or the original config when no overrides are configured
+  public static TableConfig applyConsumingOverrides(TableConfig tableConfig) {
+    if (!hasConsumingOverride(tableConfig)) {
+      return tableConfig;
+    }
+    // Enforce the same invariants that {@link #validateConsumingOverrides} enforces, so a hand-edited or
+    // older-controller-written TableConfig can't reach the merge with an unsupported override and silently produce
+    // a wrong-shape consuming segment. We re-check here (rather than relying solely on validate-time) because the
+    // merge runs on every server at consuming-segment build time and may see stale configs.
+    enforceConsumingOverrideInvariants(tableConfig);
+    try {
+      // Deep-copy the JSON tree before mutating it. {@code toJsonNode} may alias subtrees on JsonNode-typed fields
+      // of FieldConfig (notably {@code _indexes}, {@code _tierOverwrites}, {@code _consumingOverride}); mutating
+      // those in place would silently corrupt the cached TableConfig that other threads are concurrently reading.
+      JsonNode tblCfgJson = tableConfig.toJsonNode().deepCopy();
+      JsonNode fieldCfgListJson = tblCfgJson.get(TableConfig.FIELD_CONFIG_LIST_KEY);
+      // hasConsumingOverride already proved at least one FieldConfig has a non-empty override, so the JSON view
+      // must surface the field config list as an array. If not, the TableConfig (de)serialization layer is
+      // inconsistent — fail loudly rather than silently shipping the un-overridden shape.
+      Preconditions.checkState(fieldCfgListJson != null && fieldCfgListJson.isArray(),
+          "fieldConfigList missing from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      JsonNode tblIdxCfgJson = tblCfgJson.get(TableConfig.INDEXING_CONFIG_KEY);
+      Set<String> overriddenColumns = new HashSet<>();
+      Iterator<JsonNode> fieldCfgItr = fieldCfgListJson.elements();
+      while (fieldCfgItr.hasNext()) {
+        JsonNode fieldCfgJson = fieldCfgItr.next();
+        JsonNode overrideJson = fieldCfgJson.get(TableConfig.CONSUMING_OVERRIDE_KEY);
+        if (overrideJson == null || !overrideJson.isObject() || overrideJson.isEmpty()) {
+          continue;
+        }
+        JsonNode nameNode = fieldCfgJson.get("name");
+        // FieldConfig's @JsonCreator already guarantees name is non-null, but we re-parsed via toJsonNode so guard
+        // defensively. A missing name is a bug — fail loudly rather than silently dropping the override.
+        Preconditions.checkState(nameNode != null && nameNode.isTextual(),
+            "FieldConfig with consumingOverride is missing a textual 'name' field in table: %s",
+            tableConfig.getTableName());
+        overriddenColumns.add(nameNode.asText());
+        overwriteConfig(fieldCfgJson, overrideJson);
+        // Remove the override marker so downstream consumers see only the merged shape.
+        ((ObjectNode) fieldCfgJson).remove(TableConfig.CONSUMING_OVERRIDE_KEY);
+      }
+      Preconditions.checkState(!overriddenColumns.isEmpty(),
+          "no overridden columns produced from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      if (tblIdxCfgJson != null && tblIdxCfgJson.isObject()) {
+        scrubOverriddenColumnsFromTableIndexConfig((ObjectNode) tblIdxCfgJson, overriddenColumns);
+      }
+      return JsonUtils.jsonNodeToObject(tblCfgJson, TableConfig.class);
+    } catch (IOException e) {
+      // Surface override parse/deserialization failures (e.g. unknown keys, type mismatches) loudly so the bad
+      // config is rejected at validate time rather than silently falling back to the un-overridden shape — that
+      // would build the consuming segment with the wrong layout and the user would get no signal.
+      throw new IllegalStateException(
+          "Failed to apply consuming overrides for table: " + tableConfig.getTableName(), e);
+    }
+  }
+
+  /// For each top-level entry in `tableIndexConfig` that holds an array of column-name strings (legacy
+  /// per-column lists like `invertedIndexColumns`, `noDictionaryColumns`, etc.), strip any name in
+  /// `overriddenColumns`. Entries whose key is in [#CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS], that are not arrays,
+  /// or that contain non-string elements, are left alone.
+  ///
+  /// Object-shaped values (e.g. `segmentPartitionConfig.columnPartitionMap`, feature configs) are intentionally
+  /// NOT scrubbed: the heuristic "drop any object key matching an overridden column name" would silently drop
+  /// unrelated feature-config keys whose names happen to collide with a column name. Validation already rejects
+  /// overrides on partition columns up front.
+  private static void scrubOverriddenColumnsFromTableIndexConfig(ObjectNode tblIdxCfgObj,
+      Set<String> overriddenColumns) {
+    Iterator<Map.Entry<String, JsonNode>> entries = tblIdxCfgObj.properties().iterator();
+    while (entries.hasNext()) {
+      Map.Entry<String, JsonNode> entry = entries.next();
+      String key = entry.getKey();
+      if (CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS.contains(key)) {
+        continue;
+      }
+      JsonNode value = entry.getValue();
+      if (value == null || !value.isArray() || value.isEmpty()) {
+        continue;
+      }
+      ArrayNode arr = (ArrayNode) value;
+      for (JsonNode element : arr) {
+        if (!element.isTextual()) {
+          // Non-string-array entry — not a per-column list; leave alone.
+          arr = null;
+          break;
+        }
+      }
+      if (arr == null) {
+        continue;
+      }
+      List<JsonNode> kept = new ArrayList<>(arr.size());
+      for (JsonNode element : arr) {
+        if (!overriddenColumns.contains(element.asText())) {
+          kept.add(element);
+        }
+      }
+      if (kept.size() != arr.size()) {
+        ArrayNode replacement = JsonUtils.newArrayNode();
+        for (JsonNode element : kept) {
+          replacement.add(element);
+        }
+        tblIdxCfgObj.set(key, replacement);
+      }
+    }
+  }
+
+  /// Builds the [RealtimeSegmentConfig.Builder] used to construct a mutable consuming segment, applying any
+  /// [FieldConfig#getConsumingOverride()] present on the table config. When no override is configured, the
+  /// un-modified [IndexLoadingConfig]-driven builder is returned so existing code paths (tier overwrites,
+  /// instance-level mutations on `IndexLoadingConfig`) flow through unchanged. If the override merge throws (a
+  /// misconfiguration that slipped past validate), the call falls back to the `IndexLoadingConfig`-driven builder
+  /// and logs the error so consumption keeps making forward progress on the persisted shape.
+  ///
+  /// @param tableConfig source table config; not mutated
+  /// @param schema      schema in scope
+  /// @param indexLoadingConfig fallback / no-override path
+  /// @param logger      logger to emit error on fallback
+  /// @return a Builder ready for the caller to chain `.set...` calls and `.build()`
+  public static RealtimeSegmentConfig.Builder buildConsumingSegmentConfigBuilder(TableConfig tableConfig,
+      Schema schema, IndexLoadingConfig indexLoadingConfig, Logger logger) {
+    if (hasConsumingOverride(tableConfig)) {
+      try {
+        TableConfig consumingTableConfig = applyConsumingOverrides(tableConfig);
+        // Route the override-applied TableConfig back through IndexLoadingConfig so we preserve everything that
+        // IndexLoadingConfig layers on top of a raw TableConfig — instance-level config, tier overlay, schema-based
+        // timestamp-index materialization, etc. — instead of dropping them via the bare Builder(TableConfig, Schema)
+        // constructor. The IndexLoadingConfig constructor invokes TimestampIndexUtils.applyTimestampIndex which
+        // mutates its schema argument under a lock keyed to the TableConfig instance; pass a defensive schema copy
+        // so the original (possibly cluster-shared) schema is never mutated under a per-call TableConfig lock.
+        Schema schemaCopy;
+        try {
+          schemaCopy = JsonUtils.jsonNodeToObject(schema.toJsonObject(), Schema.class);
+        } catch (IOException ioe) {
+          throw new IllegalStateException("Failed to clone schema for consumingOverride path on table: "
+              + tableConfig.getTableName(), ioe);
+        }
+        IndexLoadingConfig consumingIlc = new IndexLoadingConfig(indexLoadingConfig.getInstanceDataManagerConfig(),
+            consumingTableConfig, schemaCopy);
+        String segmentTier = indexLoadingConfig.getSegmentTier();
+        if (segmentTier != null) {
+          consumingIlc.setSegmentTier(segmentTier);
+        }
+        return new RealtimeSegmentConfig.Builder(consumingIlc);
+      } catch (RuntimeException e) {
+        logger.error("Failed to apply consumingOverride for table: {}; falling back to persisted shape",
+            tableConfig.getTableName(), e);
+      }
+    }
+    return new RealtimeSegmentConfig.Builder(indexLoadingConfig);
+  }
+
+  public static boolean hasConsumingOverride(TableConfig tableConfig) {
+    List<FieldConfig> fieldConfigList = tableConfig.getFieldConfigList();
+    if (fieldConfigList == null) {
+      return false;
+    }
+    for (FieldConfig fieldConfig : fieldConfigList) {
+      JsonNode overrideJson = fieldConfig.getConsumingOverride();
+      if (overrideJson != null && overrideJson.isObject() && !overrideJson.isEmpty()) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  // Allowlist of JSON keys that {@link FieldConfig#getConsumingOverride()} may carry. Anything outside this set is
+  // either a typo or a not-yet-supported override. We reject at validate time so misconfigurations surface up front
+  // rather than getting silently dropped by Jackson via {@code @JsonIgnoreProperties(ignoreUnknown = true)} on
+  // {@link org.apache.pinot.spi.config.BaseJsonConfig}.
+  //
+  // Scope is intentionally narrow to encoding + indexing only:
+  //  - Allows {@code encodingType} (RAW vs DICTIONARY) as the high-level encoding lever.
+  //  - Allows {@code indexes} (typed per-index JSON tree, e.g. {@code {"inverted": {"enabled": true}}}) — the
+  //    canonical modern API for per-index configuration.
+  //  - Excludes the legacy {@code indexType} (singular) and {@code indexTypes} (plural) flat lists: superseded by
+  //    {@code indexes}. New features should use the typed JSON tree.
+  //  - Excludes {@code compressionCodec}: only meaningful for raw-encoded columns on disk; the consuming segment's
+  //    in-memory layout doesn't honor the on-disk codec the same way.
+  //  - Excludes {@code properties} and {@code timestampConfig}: these influence cross-subsystem state
+  //    (TransformPipeline, aggregation pipeline, timestamp index materialization) that does NOT consult the
+  //    consuming-override view, so overriding them on the consuming side would cause silent inconsistency between
+  //    the segment's index shape and the surrounding ingestion plumbing.
+  private static final Set<String> ALLOWED_CONSUMING_OVERRIDE_KEYS = Set.of("encodingType", "indexes");
+
+  /// Validates [FieldConfig#getConsumingOverride()] entries on the given table config. The override only affects
+  /// the realtime mutable consuming segment, so the rules are limited:
+  ///
+  /// - Only realtime tables make sense as targets — the override is a no-op on offline tables and is rejected to
+  ///   avoid silently misleading configurations.
+  /// - Each override JSON object must only contain keys from [#ALLOWED_CONSUMING_OVERRIDE_KEYS]; typo'd keys
+  ///   would otherwise be silently dropped by Jackson's `@JsonIgnoreProperties(ignoreUnknown = true)` on
+  ///   [org.apache.pinot.spi.config.BaseJsonConfig].
+  ///
+  /// The merged override-applied table config is also re-validated so any conflicting indexes specified inside
+  /// the override (e.g. requesting INVERTED while encoding=RAW) surface at validate time.
+  private static void validateConsumingOverrides(TableConfig tableConfig, Schema schema) {
+    if (!hasConsumingOverride(tableConfig)) {
+      return;
+    }
+    Set<String> overriddenColumns = enforceConsumingOverrideInvariants(tableConfig);
+
+    // sortedColumn is structural for the realtime consuming segment too — Builder(IndexLoadingConfig) auto-adds an
+    // inverted index to dictionary-enabled sorted columns. Allowing the override to change the sorted column's
+    // encoding/index shape would create a contradiction with that auto-handling. Reject up front.
+    List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+    if (sortedColumns != null) {
+      for (String sortedColumn : sortedColumns) {
+        Preconditions.checkState(!overriddenColumns.contains(sortedColumn),
+            "FieldConfig.consumingOverride is not allowed on sorted column: %s", sortedColumn);
+      }
+    }
+
+    // segmentPartitionConfig is keyed by column name; the consuming-side scrub would strip an overridden partition
+    // column, breaking partition routing on the consuming segment. Reject up front rather than over-scrub.
+    SegmentPartitionConfig segmentPartitionConfig = tableConfig.getIndexingConfig().getSegmentPartitionConfig();
+    if (segmentPartitionConfig != null && segmentPartitionConfig.getColumnPartitionMap() != null) {
+      for (String partitionColumn : segmentPartitionConfig.getColumnPartitionMap().keySet()) {
+        Preconditions.checkState(!overriddenColumns.contains(partitionColumn),
+            "FieldConfig.consumingOverride is not allowed on partition column: %s", partitionColumn);
+      }
+    }
+
+    // Re-run index/fieldconfig validation against the merged consuming-side shape so conflicts inside the override
+    // (e.g. requesting INVERTED on a column whose merged encoding is RAW with no dictionary) surface here. Wrap any
+    // validation error with a clear pointer back to the consumingOverride that caused it so the user sees their
+    // override (not the merged config they didn't author) in the diagnostic.
+    TableConfig merged = applyConsumingOverrides(tableConfig);
+    if (merged != tableConfig) {
+      // Defensively clone the schema before re-validating: validateIndexingConfigAndFieldConfigList may construct
+      // an IndexLoadingConfig internally, whose constructor calls TimestampIndexUtils.applyTimestampIndex which
+      // mutates the schema. Match the same defensive-copy that buildConsumingSegmentConfigBuilder applies.
+      Schema schemaCopy;
+      try {
+        schemaCopy = JsonUtils.jsonNodeToObject(schema.toJsonObject(), Schema.class);
+      } catch (IOException ioe) {
+        throw new IllegalStateException(
+            "Failed to clone schema while validating consumingOverride for table: " + tableConfig.getTableName(), ioe);
+      }
+      try {
+        validateIndexingConfigAndFieldConfigList(merged, schemaCopy);
+      } catch (RuntimeException e) {
+        throw new IllegalStateException(
+            "FieldConfig.consumingOverride for one of " + overriddenColumns
+                + " produces an invalid merged shape: " + e.getMessage(), e);
+      }
+    }
+  }
+
+  /// Shared invariant check used by both the validate path and the apply path so the rules can't drift. Verifies
+  /// the table is REALTIME and that every [FieldConfig#getConsumingOverride()] only carries keys in
+  /// [#ALLOWED_CONSUMING_OVERRIDE_KEYS]. Returns the set of column names that have a non-empty override.
+  private static Set<String> enforceConsumingOverrideInvariants(TableConfig tableConfig) {

Review Comment:
   `applyConsumingOverrides()` relies on this helper, but the helper only enforces table type and top-level keys. If a stale or hand-edited config puts `consumingOverride` on a sorted or partition column, the apply-time path will still merge it instead of rejecting/falling back, even though `validateConsumingOverrides()` explicitly forbids that state.



##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java:
##########
@@ -2111,6 +2115,318 @@ private static void overwriteConfig(JsonNode oldCfg, JsonNode newCfg) {
     }
   }
 
+  // Top-level keys in {@code tableIndexConfig} that look like per-column collections (string-array or string-keyed
+  // object) but must NOT be touched by the consuming-override scrub. Anything else of those shapes is treated as
+  // a per-column collection and the overridden column name is stripped from it. Inverting the policy this way
+  // (deny-list rather than allow-list) keeps the scrub correct as new per-column index config keys are added to
+  // {@link IndexingConfig} — there is no hand-curated list to drift out of sync.
+  //
+  // - sortedColumn is structurally tied to the segment layout; a separate validate-time check rejects overrides
+  //   on sorted columns.
+  // - starTreeIndexConfigs is a list of objects (not column names); leave it alone.
+  // - tierOverwrites is the tier-overlay JSON, handled by a separate transform.
+  private static final Set<String> CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS = Set.of(
+      "sortedColumn",
+      "starTreeIndexConfigs",
+      "tierOverwrites");
+
+  /// Returns a [TableConfig] with each [FieldConfig#getConsumingOverride()] merged into its parent
+  /// [FieldConfig]. The result is intended for building the realtime *mutable consuming* segment only.
+  /// The committed/immutable segment and all loaded immutable segments use the un-overridden table config — that is
+  /// the persisted shape on disk. A typical use is keeping the table in raw encoding for storage efficiency while
+  /// giving the consuming segment a dictionary + inverted index for fast filtering during the consuming window.
+  ///
+  /// Supported override keys: `encodingType`, `indexes`. Other [FieldConfig] fields are intentionally not
+  /// overridable — see [#ALLOWED_CONSUMING_OVERRIDE_KEYS] for the rationale.
+  ///
+  /// Merge semantics: every supported field present in `consumingOverride` replaces the parent field wholesale.
+  /// For `indexes` the entire JSON subtree is replaced rather than merged key-by-key.
+  ///
+  /// For each overridden column, the column is also scrubbed from legacy per-column lists in `tableIndexConfig`
+  /// (e.g. `invertedIndexColumns`, `rangeIndexColumns`, ...) so that those lists do not contradict the merged
+  /// [FieldConfig].
+  ///
+  /// If no [FieldConfig] declares a consuming override, the original [TableConfig] is returned unchanged.
+  ///
+  /// @param tableConfig original table config; not mutated
+  /// @return a new table config with overrides merged, or the original config when no overrides are configured
+  public static TableConfig applyConsumingOverrides(TableConfig tableConfig) {
+    if (!hasConsumingOverride(tableConfig)) {
+      return tableConfig;
+    }
+    // Enforce the same invariants that {@link #validateConsumingOverrides} enforces, so a hand-edited or
+    // older-controller-written TableConfig can't reach the merge with an unsupported override and silently produce
+    // a wrong-shape consuming segment. We re-check here (rather than relying solely on validate-time) because the
+    // merge runs on every server at consuming-segment build time and may see stale configs.
+    enforceConsumingOverrideInvariants(tableConfig);
+    try {
+      // Deep-copy the JSON tree before mutating it. {@code toJsonNode} may alias subtrees on JsonNode-typed fields
+      // of FieldConfig (notably {@code _indexes}, {@code _tierOverwrites}, {@code _consumingOverride}); mutating
+      // those in place would silently corrupt the cached TableConfig that other threads are concurrently reading.
+      JsonNode tblCfgJson = tableConfig.toJsonNode().deepCopy();
+      JsonNode fieldCfgListJson = tblCfgJson.get(TableConfig.FIELD_CONFIG_LIST_KEY);
+      // hasConsumingOverride already proved at least one FieldConfig has a non-empty override, so the JSON view
+      // must surface the field config list as an array. If not, the TableConfig (de)serialization layer is
+      // inconsistent — fail loudly rather than silently shipping the un-overridden shape.
+      Preconditions.checkState(fieldCfgListJson != null && fieldCfgListJson.isArray(),
+          "fieldConfigList missing from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      JsonNode tblIdxCfgJson = tblCfgJson.get(TableConfig.INDEXING_CONFIG_KEY);
+      Set<String> overriddenColumns = new HashSet<>();
+      Iterator<JsonNode> fieldCfgItr = fieldCfgListJson.elements();
+      while (fieldCfgItr.hasNext()) {
+        JsonNode fieldCfgJson = fieldCfgItr.next();
+        JsonNode overrideJson = fieldCfgJson.get(TableConfig.CONSUMING_OVERRIDE_KEY);
+        if (overrideJson == null || !overrideJson.isObject() || overrideJson.isEmpty()) {
+          continue;
+        }
+        JsonNode nameNode = fieldCfgJson.get("name");
+        // FieldConfig's @JsonCreator already guarantees name is non-null, but we re-parsed via toJsonNode so guard
+        // defensively. A missing name is a bug — fail loudly rather than silently dropping the override.
+        Preconditions.checkState(nameNode != null && nameNode.isTextual(),
+            "FieldConfig with consumingOverride is missing a textual 'name' field in table: %s",
+            tableConfig.getTableName());
+        overriddenColumns.add(nameNode.asText());
+        overwriteConfig(fieldCfgJson, overrideJson);
+        // Remove the override marker so downstream consumers see only the merged shape.
+        ((ObjectNode) fieldCfgJson).remove(TableConfig.CONSUMING_OVERRIDE_KEY);
+      }
+      Preconditions.checkState(!overriddenColumns.isEmpty(),
+          "no overridden columns produced from JSON view of table %s despite hasConsumingOverride==true",
+          tableConfig.getTableName());
+      if (tblIdxCfgJson != null && tblIdxCfgJson.isObject()) {
+        scrubOverriddenColumnsFromTableIndexConfig((ObjectNode) tblIdxCfgJson, overriddenColumns);
+      }
+      return JsonUtils.jsonNodeToObject(tblCfgJson, TableConfig.class);
+    } catch (IOException e) {
+      // Surface override parse/deserialization failures (e.g. unknown keys, type mismatches) loudly so the bad
+      // config is rejected at validate time rather than silently falling back to the un-overridden shape — that
+      // would build the consuming segment with the wrong layout and the user would get no signal.
+      throw new IllegalStateException(
+          "Failed to apply consuming overrides for table: " + tableConfig.getTableName(), e);
+    }
+  }
+
+  /// For each top-level entry in `tableIndexConfig` that holds an array of column-name strings (legacy
+  /// per-column lists like `invertedIndexColumns`, `noDictionaryColumns`, etc.), strip any name in
+  /// `overriddenColumns`. Entries whose key is in [#CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS], that are not arrays,
+  /// or that contain non-string elements, are left alone.
+  ///
+  /// Object-shaped values (e.g. `segmentPartitionConfig.columnPartitionMap`, feature configs) are intentionally
+  /// NOT scrubbed: the heuristic "drop any object key matching an overridden column name" would silently drop
+  /// unrelated feature-config keys whose names happen to collide with a column name. Validation already rejects
+  /// overrides on partition columns up front.
+  private static void scrubOverriddenColumnsFromTableIndexConfig(ObjectNode tblIdxCfgObj,
+      Set<String> overriddenColumns) {
+    Iterator<Map.Entry<String, JsonNode>> entries = tblIdxCfgObj.properties().iterator();
+    while (entries.hasNext()) {
+      Map.Entry<String, JsonNode> entry = entries.next();
+      String key = entry.getKey();
+      if (CONSUMING_OVERRIDE_SCRUB_EXCLUDED_KEYS.contains(key)) {
+        continue;
+      }
+      JsonNode value = entry.getValue();
+      if (value == null || !value.isArray() || value.isEmpty()) {
+        continue;
+      }
+      ArrayNode arr = (ArrayNode) value;
+      for (JsonNode element : arr) {
+        if (!element.isTextual()) {
+          // Non-string-array entry — not a per-column list; leave alone.
+          arr = null;
+          break;
+        }
+      }
+      if (arr == null) {
+        continue;
+      }
+      List<JsonNode> kept = new ArrayList<>(arr.size());
+      for (JsonNode element : arr) {
+        if (!overriddenColumns.contains(element.asText())) {
+          kept.add(element);
+        }
+      }
+      if (kept.size() != arr.size()) {
+        ArrayNode replacement = JsonUtils.newArrayNode();
+        for (JsonNode element : kept) {
+          replacement.add(element);
+        }
+        tblIdxCfgObj.set(key, replacement);
+      }
+    }
+  }
+
+  /// Builds the [RealtimeSegmentConfig.Builder] used to construct a mutable consuming segment, applying any
+  /// [FieldConfig#getConsumingOverride()] present on the table config. When no override is configured, the
+  /// un-modified [IndexLoadingConfig]-driven builder is returned so existing code paths (tier overwrites,
+  /// instance-level mutations on `IndexLoadingConfig`) flow through unchanged. If the override merge throws (a
+  /// misconfiguration that slipped past validate), the call falls back to the `IndexLoadingConfig`-driven builder
+  /// and logs the error so consumption keeps making forward progress on the persisted shape.
+  ///
+  /// @param tableConfig source table config; not mutated
+  /// @param schema      schema in scope
+  /// @param indexLoadingConfig fallback / no-override path
+  /// @param logger      logger to emit error on fallback
+  /// @return a Builder ready for the caller to chain `.set...` calls and `.build()`
+  public static RealtimeSegmentConfig.Builder buildConsumingSegmentConfigBuilder(TableConfig tableConfig,
+      Schema schema, IndexLoadingConfig indexLoadingConfig, Logger logger) {
+    if (hasConsumingOverride(tableConfig)) {
+      try {
+        TableConfig consumingTableConfig = applyConsumingOverrides(tableConfig);
+        // Route the override-applied TableConfig back through IndexLoadingConfig so we preserve everything that
+        // IndexLoadingConfig layers on top of a raw TableConfig — instance-level config, tier overlay, schema-based
+        // timestamp-index materialization, etc. — instead of dropping them via the bare Builder(TableConfig, Schema)
+        // constructor. The IndexLoadingConfig constructor invokes TimestampIndexUtils.applyTimestampIndex which
+        // mutates its schema argument under a lock keyed to the TableConfig instance; pass a defensive schema copy
+        // so the original (possibly cluster-shared) schema is never mutated under a per-call TableConfig lock.
+        Schema schemaCopy;
+        try {
+          schemaCopy = JsonUtils.jsonNodeToObject(schema.toJsonObject(), Schema.class);
+        } catch (IOException ioe) {
+          throw new IllegalStateException("Failed to clone schema for consumingOverride path on table: "
+              + tableConfig.getTableName(), ioe);
+        }
+        IndexLoadingConfig consumingIlc = new IndexLoadingConfig(indexLoadingConfig.getInstanceDataManagerConfig(),
+            consumingTableConfig, schemaCopy);
+        String segmentTier = indexLoadingConfig.getSegmentTier();
+        if (segmentTier != null) {
+          consumingIlc.setSegmentTier(segmentTier);
+        }
+        return new RealtimeSegmentConfig.Builder(consumingIlc);
+      } catch (RuntimeException e) {
+        logger.error("Failed to apply consumingOverride for table: {}; falling back to persisted shape",
+            tableConfig.getTableName(), e);
+      }
+    }
+    return new RealtimeSegmentConfig.Builder(indexLoadingConfig);
+  }
+
+  public static boolean hasConsumingOverride(TableConfig tableConfig) {
+    List<FieldConfig> fieldConfigList = tableConfig.getFieldConfigList();
+    if (fieldConfigList == null) {
+      return false;
+    }
+    for (FieldConfig fieldConfig : fieldConfigList) {
+      JsonNode overrideJson = fieldConfig.getConsumingOverride();
+      if (overrideJson != null && overrideJson.isObject() && !overrideJson.isEmpty()) {

Review Comment:
   A malformed `consumingOverride` value such as `[]` or `"foo"` is treated as if the field were absent here, so both validation and apply-time logic silently ignore it instead of surfacing a config error. Because the field is a raw `JsonNode`, we need an explicit type check/rejection for non-object values to avoid unnoticed misconfigurations.
   



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org