Add a StrictTypeIdResolver which would throw an exception if type is not in the predefined json subtype list. Use it in DimensionSchema.

processing/src/main/java/org/apache/druid/data/input/impl/DimensionSchema.java

@@ -25,9 +25,11 @@
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.annotation.JsonValue;
+import com.fasterxml.jackson.databind.annotation.JsonTypeResolver;
 import com.google.common.base.Strings;
 import org.apache.druid.guice.BuiltInTypesModule;
 import org.apache.druid.guice.annotations.PublicApi;
+import org.apache.druid.jackson.StrictTypeIdResolver;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.emitter.EmittingLogger;
 import org.apache.druid.segment.AutoTypeColumnSchema;
@@ -44,9 +46,27 @@
 import java.util.Objects;
 
 /**
+ * Defines the schema of a single dimension in a dataset.
+ * <p>
+ * Includes metadata such as the dimension's name, type, and whether it
+ * can hold multiple values. Supports Jackson serialization/deserialization,
+ * including polymorphic types via {@code @JsonSubTypes}.
+ * </p>
+ *
+ * <p>
+ * Example JSON:
+ * <pre>{@code
+ * {
+ *     "type": "string",
+ *     "name": "country",
+ *     "multiValue": false
+ * }
+ * }</pre>
+ * </p>
  */
 @PublicApi
-@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type", defaultImpl = StringDimensionSchema.class)
+@JsonTypeResolver(StrictTypeIdResolver.Builder.class)
+@JsonTypeInfo(use = JsonTypeInfo.Id.CUSTOM, property = "type", defaultImpl = StringDimensionSchema.class)
 @JsonSubTypes(value = {
     @JsonSubTypes.Type(name = DimensionSchema.STRING_TYPE_NAME, value = StringDimensionSchema.class),
     @JsonSubTypes.Type(name = DimensionSchema.LONG_TYPE_NAME, value = LongDimensionSchema.class),

processing/src/main/java/org/apache/druid/jackson/StrictTypeIdResolver.java

@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.jackson;
+
+import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DatabindContext;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.cfg.MapperConfig;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
+import com.fasterxml.jackson.databind.jsontype.PolymorphicTypeValidator;
+import com.fasterxml.jackson.databind.jsontype.TypeIdResolver;
+import com.fasterxml.jackson.databind.jsontype.impl.StdTypeResolverBuilder;
+import com.fasterxml.jackson.databind.jsontype.impl.TypeIdResolverBase;
+import com.fasterxml.jackson.databind.jsontype.impl.TypeNameIdResolver;
+
+import java.util.Collection;
+
+/**
+ * A strict {@link TypeIdResolver} implementation that validates all incoming type ids.
+ * <p>
+ * During deserialization, the type discriminator in the JSON must correspond to a registered subtype;
+ * otherwise this resolver will throw an exception instead of silently accepting or defaulting.
+ * <p>
+ * An optional default implementation may still be configured and used only when the type id is absent.
+ */
+public class StrictTypeIdResolver extends TypeIdResolverBase
+{
+  public static class Builder extends StdTypeResolverBuilder
+  {
+    @Override
+    protected TypeIdResolver idResolver(
+        MapperConfig<?> config,
+        JavaType baseType,
+        PolymorphicTypeValidator subtypeValidator,
+        Collection<NamedType> subtypes,
+        boolean forSer,
+        boolean forDeser
+    )
+    {
+      this._customIdResolver = new StrictTypeIdResolver(config, baseType, subtypes, forSer, forDeser);
+      return this._customIdResolver;
+    }
+  }
+
+  protected final JavaType baseType;
+  protected final TypeNameIdResolver delegate;
+
+  StrictTypeIdResolver()
+  {
+    // Required default constructor for Jackson, the instance is never used
+    baseType = null;
+    delegate = null;
+  }
+
+  StrictTypeIdResolver(
+      MapperConfig<?> config,
+      JavaType baseType,
+      Collection<NamedType> subtypes,
+      boolean forSer,
+      boolean forDeser
+  )
+  {
+    this.baseType = baseType;
+    this.delegate = TypeNameIdResolver.construct(config, baseType, subtypes, forSer, forDeser);
+  }
+
+  @Override
+  public JavaType typeFromId(DatabindContext context, String id) throws JsonProcessingException
+  {
+    JavaType type = delegate.typeFromId(context, id);
+    if (type == null) {
+      // in TypeNameIdResolver, it'd fall back to defaultImpl if configured, but we want to error out instead
+      throw ((DeserializationContext) context).invalidTypeIdException(
+          baseType,
+          id,
+          "known type ids = " + delegate.getDescForKnownTypeIds()
+      );
+    }
+    return type;
+  }
+
+  @Override
+  public String idFromValue(Object value)
+  {
+    return delegate.idFromValue(value);
+  }
+
+  @Override
+  public String idFromValueAndType(Object value, Class<?> suggestedType)
+  {
+    return delegate.idFromValueAndType(value, suggestedType);
+  }
+
+  @Override
+  public String getDescForKnownTypeIds()
+  {
+    return delegate.getDescForKnownTypeIds();
+  }
+
+  @Override
+  public Id getMechanism()
+  {
+    return Id.CUSTOM;
+  }
+}

processing/src/test/java/org/apache/druid/data/input/impl/DimensionSchemaTest.java

@@ -20,6 +20,7 @@
 package org.apache.druid.data.input.impl;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.exc.InvalidTypeIdException;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -46,4 +47,26 @@ public void testStringDimensionSchemaSerde() throws Exception
         OBJECT_MAPPER.readValue(OBJECT_MAPPER.writeValueAsString(schema2), DimensionSchema.class)
     );
   }
+
+  @Test
+  public void testDeserializeStrictTypeId() throws Exception
+  {
+    final String invalidType = "{\"type\":\"invalid\",\"name\":\"foo\",\"multiValueHandling\":\"ARRAY\",\"createBitmapIndex\":false}";
+    InvalidTypeIdException e = Assert.assertThrows(
+        InvalidTypeIdException.class,
+        () -> OBJECT_MAPPER.readValue(invalidType, DimensionSchema.class)
+    );
+    Assert.assertTrue(e.getMessage().contains("Could not resolve type id"));
+    Assert.assertTrue(e.getMessage().contains("invalid"));
+  }
+
+  @Test
+  public void testDeserializeDefaultAsString() throws Exception
+  {
+    final String noType = "{\"name\":\"foo\",\"multiValueHandling\":\"ARRAY\",\"createBitmapIndex\":false}";
+    Assert.assertEquals(
+        new StringDimensionSchema("foo", DimensionSchema.MultiValueHandling.ARRAY, false),
+        OBJECT_MAPPER.readValue(noType, DimensionSchema.class)
+    );
+  }
 }

processing/src/test/java/org/apache/druid/data/input/impl/StringDimensionSchemaTest.java

@@ -52,7 +52,6 @@ public void testDeserializeFromSimpleString() throws JsonProcessingException
   public void testDeserializeFromJson() throws JsonProcessingException
   {
     final String json = "{\n"
-                        + "  \"type\" : \"StringDimensionSchema\",\n"
                         + "  \"name\" : \"dim\",\n"
                         + "  \"multiValueHandling\" : \"SORTED_SET\",\n"
                         + "  \"createBitmapIndex\" : false\n"