Annotation-driven deprecated table config validation on create + update

pinot-clients/pinot-java-client/src/main/java/org/apache/pinot/client/admin/README.md

@@ -147,6 +147,68 @@ String status = tableClient.getTableStatus("myTable");
 String rebalanceResult = tableClient.rebalanceTable("myTable", true, "default", 1);
 ```
 
+### Creating Tables With Modern Config Fields
+
+Table creation endpoints reject deprecated table-config properties when they are explicitly present in the request
+payload. Existing tables can still be updated with legacy fields for backward compatibility, but new create requests
+must use the current config layout.
+
+Common migrations:
+
+- `segmentsConfig.segmentPushType` -> `ingestionConfig.batchIngestionConfig.segmentIngestionType`
+- `segmentsConfig.segmentPushFrequency` -> `ingestionConfig.batchIngestionConfig.segmentIngestionFrequency`
+- `tableIndexConfig.streamConfigs` -> `ingestionConfig.streamIngestionConfig.streamConfigMaps`
+- `fieldConfigList[].indexType` -> `fieldConfigList[].indexTypes`
+
+Sample REALTIME table config for create:
+
+```json
+{
+  "tableName": "events",
+  "tableType": "REALTIME",
+  "segmentsConfig": {
+    "timeColumnName": "ts",
+    "replication": "1"
+  },
+  "tenants": {
+    "broker": "DefaultTenant",
+    "server": "DefaultTenant"
+  },
+  "tableIndexConfig": {
+    "loadMode": "MMAP"
+  },
+  "fieldConfigList": [
+    {
+      "name": "userId",
+      "encodingType": "DICTIONARY",
+      "indexTypes": [
+        "INVERTED"
+      ]
+    }
+  ],
+  "ingestionConfig": {
+    "batchIngestionConfig": {
+      "segmentIngestionType": "APPEND",
+      "segmentIngestionFrequency": "DAILY"
+    },
+    "streamIngestionConfig": {
+      "streamConfigMaps": [
+        {
+          "streamType": "kafka",
+          "stream.kafka.topic.name": "events",
+          "stream.kafka.consumer.type": "lowlevel",
+          "stream.kafka.decoder.class.name": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder"
+        }
+      ]
+    }
+  },
+  "metadata": {}
+}
+```
+
+If a create request still sends deprecated fields, the controller returns `400 BAD_REQUEST` with the offending JSON path
+and the replacement field to use.
+
 ### Schema Operations
 
 ```java

pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java

@@ -519,6 +519,16 @@ public static TableConfig getTableConfig(ZkHelixPropertyStore<ZNRecord> property
         AccessOption.PERSISTENT), replaceVariables, applyDecorator);
   }
 
+  /// Returns the raw [ZNRecord] for a stored table config without deserializing it. Use this when callers need the
+  /// byte-faithful view of what was last written to ZK (e.g. update-time deprecation diffing). Returns null when
+  /// the table does not exist.
+  @Nullable
+  public static ZNRecord getTableConfigZNRecord(ZkHelixPropertyStore<ZNRecord> propertyStore,
+      String tableNameWithType) {
+    return propertyStore.get(constructPropertyStorePathForResourceConfig(tableNameWithType), null,
+        AccessOption.PERSISTENT);
+  }
+
   @Nullable
   public static ImmutablePair<TableConfig, Stat> getTableConfigWithStat(ZkHelixPropertyStore<ZNRecord> propertyStore,
       String tableNameWithType) {

pinot-common/src/main/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtils.java

@@ -20,6 +20,9 @@
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.JsonNodeFactory;
+import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.base.Preconditions;
 import java.io.IOException;
 import java.util.HashMap;
@@ -293,4 +296,36 @@ public static ZNRecord toZNRecord(TableConfig tableConfig)
     znRecord.setSimpleFields(simpleFields);
     return znRecord;
   }
+
+  /// Reconstructs the table-config JSON tree from a stored [ZNRecord], preserving every key originally written to
+  /// ZK. Unlike `TableConfig.toJsonNode()`, this method does not round-trip through the Java bean and therefore
+  /// does not strip fields that the bean's getters mark with `@JsonIgnore` or `@JsonInclude(NON_DEFAULT)`. It is
+  /// intended for code paths (e.g. update-time deprecation diffing) that need to compare an incoming request
+  /// against the exact bytes that were stored.
+  ///
+  /// @param znRecord the raw ZNRecord read from the property store
+  /// @return a JsonNode equivalent to what the user originally PUT/POST-ed for the table, or `null` if the input is
+  /// `null`
+  public static JsonNode toRawJsonNode(ZNRecord znRecord) {
+    if (znRecord == null) {
+      return null;
+    }
+    ObjectNode root = JsonNodeFactory.instance.objectNode();
+    Map<String, String> simpleFields = znRecord.getSimpleFields();
+    for (Map.Entry<String, String> entry : simpleFields.entrySet()) {
+      String key = entry.getKey();
+      String value = entry.getValue();
+      // Try to parse as a JSON node first (most fields hold JSON-encoded sub-objects). Fall back to a string node
+      // for primitive simple fields (e.g. tableType, isDimTable).
+      try {
+        root.set(key, JsonUtils.stringToJsonNode(value));
+      } catch (IOException e) {
+        root.put(key, value);
+      }
+    }
+    if (!root.has(TableConfig.TABLE_NAME_KEY)) {
+      root.put(TableConfig.TABLE_NAME_KEY, znRecord.getId());
+    }
+    return root;
+  }
 }

pinot-common/src/test/java/org/apache/pinot/common/utils/config/TableConfigSerDeUtilsTest.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pinot.common.utils.config;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.collect.Lists;
 import java.io.IOException;
@@ -26,6 +27,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
 import org.apache.pinot.common.tier.TierFactory;
 import org.apache.pinot.spi.config.table.CompletionConfig;
 import org.apache.pinot.spi.config.table.DedupConfig;
@@ -607,4 +609,31 @@ private void checkTableConfigWithDedupConfigWithTTL(TableConfig tableConfig) {
     assertEquals(dedupConfig.getMetadataTTL(), 10);
     assertEquals(dedupConfig.getDedupTimeColumn(), "dedupTimeColumn");
   }
+
+  /**
+   * Confirms that {@code toRawJsonNode} preserves keys that are stripped by the deserialize/re-serialize round-trip
+   * (e.g. {@code @JsonIgnore} or {@code @JsonInclude(NON_DEFAULT)} getters). The deprecation diff in
+   * {@code DeprecatedTableConfigValidationUtils} relies on this.
+   */
+  @Test
+  public void testToRawJsonNodePreservesStrippedKeys() {
+    ZNRecord znRecord = new ZNRecord("myTable_OFFLINE");
+    znRecord.setSimpleField(TableConfig.TABLE_NAME_KEY, "myTable_OFFLINE");
+    znRecord.setSimpleField(TableConfig.TABLE_TYPE_KEY, "OFFLINE");
+    // The fieldConfigList stored in ZK still contains the legacy 'indexType' key (written before @JsonIgnore).
+    znRecord.setSimpleField(TableConfig.FIELD_CONFIG_LIST_KEY,
+        "[{\"name\":\"c1\",\"indexType\":\"INVERTED\",\"indexTypes\":[\"INVERTED\"]}]");
+
+    JsonNode raw = TableConfigSerDeUtils.toRawJsonNode(znRecord);
+    assertNotNull(raw);
+    JsonNode legacy = raw.get(TableConfig.FIELD_CONFIG_LIST_KEY);
+    assertNotNull(legacy);
+    assertTrue(legacy.isArray());
+    assertEquals(legacy.get(0).get("indexType").asText(), "INVERTED");
+  }
+
+  @Test
+  public void testToRawJsonNodeHandlesNull() {
+    assertNull(TableConfigSerDeUtils.toRawJsonNode(null));
+  }
 }

pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/ConfigSuccessResponse.java

@@ -18,17 +18,33 @@
  */
 package org.apache.pinot.controller.api.resources;
 
+import com.fasterxml.jackson.annotation.JsonInclude;
+import java.util.List;
 import java.util.Map;
 
 public final class ConfigSuccessResponse extends SuccessResponse {
   private final Map<String, Object> _unrecognizedProperties;
+  private final List<String> _deprecationWarnings;
 
   public ConfigSuccessResponse(String status, Map<String, Object> unrecognizedProperties) {
+    this(status, unrecognizedProperties, List.of());
+  }
+
+  public ConfigSuccessResponse(String status, Map<String, Object> unrecognizedProperties,
+      List<String> deprecationWarnings) {
     super(status);
-    _unrecognizedProperties = unrecognizedProperties;
+    _unrecognizedProperties = unrecognizedProperties == null ? Map.of() : unrecognizedProperties;
+    _deprecationWarnings = deprecationWarnings == null ? List.of() : deprecationWarnings;
   }
 
   public Map<String, Object> getUnrecognizedProperties() {
     return _unrecognizedProperties;
   }
+
+  /// `@JsonInclude(NON_EMPTY)` is on the getter so the empty-list case is elided from the response. Older clients
+  /// that strict-parse this DTO continue to see the original (pre-deprecationWarnings) shape when no warnings fire.
+  @JsonInclude(JsonInclude.Include.NON_EMPTY)
+  public List<String> getDeprecationWarnings() {
+    return _deprecationWarnings;
+  }
 }