diff --git a/docs/multi-stage-query/reference.md b/docs/multi-stage-query/reference.md
index 32c38a60abd..71fc1b43afe 100644
--- a/docs/multi-stage-query/reference.md
+++ b/docs/multi-stage-query/reference.md
@@ -402,6 +402,13 @@ FROM TABLE(
   ) (x VARCHAR, y VARCHAR)
 ```
 
+The JSON function allows columns to be of type `TYPE('COMPLEX<json>')` which indicates that the column contains
+some form of complex JSON: a JSON object, a JSON array, or an array of JSON objects or arrays.
+Note that the case must exactly match that given: upper case `COMPLEX`, lower case `json`.
+The SQL type simply names a native Druid type. However, the actual
+segment column produced may be of some other type if Druid infers that it can use a simpler type
+instead.
+
 ### Parameters
 
 Starting with the Druid 26.0 release, you can use query parameters with MSQ queries. You may find
diff --git a/server/src/main/java/org/apache/druid/catalog/model/Columns.java b/server/src/main/java/org/apache/druid/catalog/model/Columns.java
index 05820086f69..0bc0f71bc1b 100644
--- a/server/src/main/java/org/apache/druid/catalog/model/Columns.java
+++ b/server/src/main/java/org/apache/druid/catalog/model/Columns.java
@@ -81,7 +81,11 @@ public class Columns
     if (sqlType == null) {
       return null;
     }
-    return SQL_TO_DRUID_TYPES.get(StringUtils.toUpperCase(sqlType));
+    ColumnType druidType = SQL_TO_DRUID_TYPES.get(StringUtils.toUpperCase(sqlType));
+    if (druidType != null) {
+      return druidType;
+    }
+    return ColumnType.fromString(sqlType);
   }
 
   public static void validateScalarColumn(String name, String type)
@@ -105,7 +109,7 @@ public class Columns
     for (ColumnSpec col : columns) {
       ColumnType druidType = null;
       if (col.sqlType() != null) {
-        druidType = Columns.SQL_TO_DRUID_TYPES.get(StringUtils.toUpperCase(col.sqlType()));
+        druidType = Columns.druidType(col.sqlType());
       }
       if (druidType == null) {
         druidType = ColumnType.STRING;
diff --git a/sql/src/main/codegen/config.fmpp b/sql/src/main/codegen/config.fmpp
index fce1ec2f3a2..b0f49eebc90 100644
--- a/sql/src/main/codegen/config.fmpp
+++ b/sql/src/main/codegen/config.fmpp
@@ -402,6 +402,7 @@ data: {
     # Return type of method implementation should be "SqlTypeNameSpec".
     # Example: SqlParseTimeStampZ().
     dataTypeParserMethods: [
+      "DruidType()"
     ]
 
     # List of methods for parsing builtin function calls.
diff --git a/sql/src/main/codegen/includes/common.ftl b/sql/src/main/codegen/includes/common.ftl
index 95ad1b509c4..2eccdbc2a5a 100644
--- a/sql/src/main/codegen/includes/common.ftl
+++ b/sql/src/main/codegen/includes/common.ftl
@@ -92,3 +92,18 @@ SqlNodeList ClusterItems() :
     return new SqlNodeList(list, s.addAll(list).pos());
   }
 }
+
+SqlTypeNameSpec DruidType() :
+{
+  String typeName;
+}
+{
+  <TYPE> <LPAREN> <QUOTED_STRING>
+  {
+    typeName = SqlParserUtil.trim(token.image, "'");
+  }
+  <RPAREN>
+  {
+    return new SqlUserDefinedTypeNameSpec(typeName, span().pos());
+  }
+}
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/external/Externals.java b/sql/src/main/java/org/apache/druid/sql/calcite/external/Externals.java
index ae15d1b45b1..db1d38618d6 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/external/Externals.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/external/Externals.java
@@ -238,11 +238,15 @@ public class Externals
     if (spec == null) {
       throw unsupportedType(name, dataType);
     }
-    SqlIdentifier typeName = spec.getTypeName();
-    if (typeName == null || !typeName.isSimple()) {
+    SqlIdentifier typeNameIdentifier = spec.getTypeName();
+    if (typeNameIdentifier == null || !typeNameIdentifier.isSimple()) {
       throw unsupportedType(name, dataType);
     }
-    SqlTypeName type = SqlTypeName.get(typeName.getSimple());
+    String simpleName = typeNameIdentifier.getSimple();
+    if (StringUtils.toLowerCase(simpleName).startsWith(("complex<"))) {
+      return simpleName;
+    }
+    SqlTypeName type = SqlTypeName.get(simpleName);
     if (type == null) {
       throw unsupportedType(name, dataType);
     }
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidTypeSystem.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidTypeSystem.java
index 1279667ad65..160920c6208 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidTypeSystem.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidTypeSystem.java
@@ -35,6 +35,8 @@ public class DruidTypeSystem implements RelDataTypeSystem
    */
   public static final int DEFAULT_TIMESTAMP_PRECISION = 3;
 
+  public static final String VARIANT_TYPE_NAME = "VARIANT";
+
   private DruidTypeSystem()
   {
     // Singleton.
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/IngestTableFunctionTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/IngestTableFunctionTest.java
index f23a1353591..4a6b4445598 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/IngestTableFunctionTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/IngestTableFunctionTest.java
@@ -34,6 +34,7 @@ import org.apache.druid.java.util.common.UOE;
 import org.apache.druid.metadata.DefaultPasswordProvider;
 import org.apache.druid.segment.column.ColumnType;
 import org.apache.druid.segment.column.RowSignature;
+import org.apache.druid.segment.nested.NestedDataComplexTypeSerde;
 import org.apache.druid.sql.calcite.external.ExternalDataSource;
 import org.apache.druid.sql.calcite.external.Externals;
 import org.apache.druid.sql.calcite.filtration.Filtration;
@@ -207,6 +208,45 @@ public class IngestTableFunctionTest extends CalciteIngestionDmlTest
         .verify();
   }
 
+  @Test
+  public void testHttpJson()
+  {
+    final ExternalDataSource httpDataSource = new ExternalDataSource(
+        new HttpInputSource(
+            Collections.singletonList(toURI("http://foo.com/bar.json")),
+            "bob",
+            new DefaultPasswordProvider("secret"),
+            new HttpInputSourceConfig(null)
+        ),
+        new CsvInputFormat(ImmutableList.of("x", "y", "z"), null, false, false, 0),
+        RowSignature.builder()
+                    .add("x", ColumnType.STRING)
+                    .add("y", ColumnType.STRING)
+                    .add("z", NestedDataComplexTypeSerde.TYPE)
+                    .build()
+        );
+    testIngestionQuery()
+        .sql("INSERT INTO dst SELECT *\n" +
+             "FROM TABLE(http(userName => 'bob',\n" +
+            "                 password => 'secret',\n" +
+             "                uris => ARRAY['http://foo.com/bar.json'],\n" +
+             "                format => 'csv'))\n" +
+             "     EXTEND (x VARCHAR, y VARCHAR, z TYPE('COMPLEX<json>'))\n" +
+             "PARTITIONED BY ALL TIME")
+        .authentication(CalciteTests.SUPER_USER_AUTH_RESULT)
+        .expectTarget("dst", httpDataSource.getSignature())
+        .expectResources(dataSourceWrite("dst"), Externals.EXTERNAL_RESOURCE_ACTION)
+        .expectQuery(
+            newScanQueryBuilder()
+                .dataSource(httpDataSource)
+                .intervals(querySegmentSpec(Filtration.eternity()))
+                .columns("x", "y", "z")
+                .context(CalciteIngestionDmlTest.PARTITIONED_BY_ALL_TIME_QUERY_CONTEXT)
+                .build()
+         )
+        .verify();
+  }
+
   /**
    * Basic use of an inline input source via EXTERN
    */