diff --git a/core/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java b/core/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java
index e06edc0598f7..29f20a8f3280 100644
--- a/core/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java
+++ b/core/src/main/java/org/apache/calcite/sql/type/SqlTypeUtil.java
@@ -21,13 +21,18 @@
 import org.apache.calcite.rel.type.RelDataTypeFamily;
 import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rel.type.RelDataTypeFieldImpl;
+import org.apache.calcite.rel.type.RelRecordType;
 import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.sql.SqlBasicTypeNameSpec;
 import org.apache.calcite.sql.SqlCall;
 import org.apache.calcite.sql.SqlCallBinding;
 import org.apache.calcite.sql.SqlCollation;
+import org.apache.calcite.sql.SqlCollectionTypeNameSpec;
 import org.apache.calcite.sql.SqlDataTypeSpec;
+import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlRowTypeNameSpec;
+import org.apache.calcite.sql.SqlTypeNameSpec;
 import org.apache.calcite.sql.parser.SqlParserPos;
 import org.apache.calcite.sql.validate.SqlNameMatcher;
 import org.apache.calcite.sql.validate.SqlValidator;
@@ -1038,19 +1043,40 @@ public static SqlDataTypeSpec convertTypeToSpec(RelDataType type,
     // interval types, multiset types, etc
     assert typeName != null;
 
-    int precision = typeName.allowsPrec() ? type.getPrecision() : -1;
-    // fix up the precision.
-    if (maxPrecision > 0 && precision > maxPrecision) {
-      precision = maxPrecision;
+    final SqlTypeNameSpec typeNameSpec;
+    if (isAtomic(type)) {
+      int precision = typeName.allowsPrec() ? type.getPrecision() : -1;
+      // fix up the precision.
+      if (maxPrecision > 0 && precision > maxPrecision) {
+        precision = maxPrecision;
+      }
+      int scale = typeName.allowsScale() ? type.getScale() : -1;
+
+      typeNameSpec = new SqlBasicTypeNameSpec(
+          typeName,
+          precision,
+          scale,
+          charSetName,
+          SqlParserPos.ZERO);
+    } else if (isCollection(type)) {
+      typeNameSpec = new SqlCollectionTypeNameSpec(
+          convertTypeToSpec(type.getComponentType()).getTypeNameSpec(),
+          typeName,
+          SqlParserPos.ZERO);
+    } else if (isRow(type)) {
+      RelRecordType recordType = (RelRecordType) type;
+      List<RelDataTypeField> fields = recordType.getFieldList();
+      List<SqlIdentifier> fieldNames = fields.stream()
+          .map(f -> new SqlIdentifier(f.getName(), SqlParserPos.ZERO))
+          .collect(Collectors.toList());
+      List<SqlDataTypeSpec> fieldTypes = fields.stream()
+          .map(f -> convertTypeToSpec(f.getType()))
+          .collect(Collectors.toList());
+      typeNameSpec = new SqlRowTypeNameSpec(SqlParserPos.ZERO, fieldNames, fieldTypes);
+    } else {
+      throw new UnsupportedOperationException(
+          "Unsupported type when convertTypeToSpec: " + typeName);
     }
-    int scale = typeName.allowsScale() ? type.getScale() : -1;
-
-    final SqlBasicTypeNameSpec typeNameSpec = new SqlBasicTypeNameSpec(
-        typeName,
-        precision,
-        scale,
-        charSetName,
-        SqlParserPos.ZERO);
 
     // REVIEW jvs 28-Dec-2004:  discriminate between precision/scale
     // zero and unspecified?
@@ -1498,6 +1524,26 @@ public static boolean isArray(RelDataType type) {
     return type.getSqlTypeName() == SqlTypeName.ARRAY;
   }
 
+  /**
+   * @return true if type is COLLECTION
+   */
+  public static boolean isCollection(RelDataType type) {
+    return type.getSqlTypeName() == SqlTypeName.ARRAY
+        || type.getSqlTypeName() == SqlTypeName.MULTISET;
+  }
+
+  /**
+   * @return true if type is ROW
+   */
+  public static boolean isRow(RelDataType type) {
+    SqlTypeName typeName = type.getSqlTypeName();
+    if (typeName == null) {
+      return false;
+    }
+    return type.getSqlTypeName() == SqlTypeName.ROW;
+  }
+
+
   /**
    * @return true if type is MAP
    */
diff --git a/core/src/test/java/org/apache/calcite/sql/type/SqlTypeUtilTest.java b/core/src/test/java/org/apache/calcite/sql/type/SqlTypeUtilTest.java
index 61658113d2d4..e1fdbddbcafb 100644
--- a/core/src/test/java/org/apache/calcite/sql/type/SqlTypeUtilTest.java
+++ b/core/src/test/java/org/apache/calcite/sql/type/SqlTypeUtilTest.java
@@ -18,12 +18,21 @@
 
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.sql.SqlBasicTypeNameSpec;
+import org.apache.calcite.sql.SqlCollectionTypeNameSpec;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlRowTypeNameSpec;
 
 import com.google.common.collect.ImmutableList;
 
 import org.junit.jupiter.api.Test;
 
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
 import static org.apache.calcite.sql.type.SqlTypeUtil.areSameFamily;
+import static org.apache.calcite.sql.type.SqlTypeUtil.convertTypeToSpec;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.MatcherAssert.assertThat;
@@ -117,6 +126,36 @@ class SqlTypeUtilTest {
     SqlTypeCoercionRule.THREAD_PROVIDERS.set(defaultRules);
   }
 
+  @Test void testConvertTypeToSpec() {
+    SqlBasicTypeNameSpec basicSpec =
+        (SqlBasicTypeNameSpec) convertTypeToSpec(f.sqlBigInt).getTypeNameSpec();
+    assertThat(basicSpec.getTypeName().getSimple(), is("BIGINT"));
+
+    SqlCollectionTypeNameSpec arraySpec =
+        (SqlCollectionTypeNameSpec) convertTypeToSpec(f.arrayBigInt).getTypeNameSpec();
+    assertThat(arraySpec.getTypeName().getSimple(), is("ARRAY"));
+    assertThat(arraySpec.getElementTypeName().getTypeName().getSimple(), is("BIGINT"));
+
+    SqlCollectionTypeNameSpec multisetSpec =
+        (SqlCollectionTypeNameSpec) convertTypeToSpec(f.multisetBigInt).getTypeNameSpec();
+    assertThat(multisetSpec.getTypeName().getSimple(), is("MULTISET"));
+    assertThat(multisetSpec.getElementTypeName().getTypeName().getSimple(), is("BIGINT"));
+
+    SqlRowTypeNameSpec rowSpec =
+        (SqlRowTypeNameSpec) convertTypeToSpec(f.structOfInt).getTypeNameSpec();
+    List<String> fieldNames = rowSpec.getFieldNames()
+        .stream()
+        .map(SqlIdentifier::getSimple)
+        .collect(Collectors.toList());
+    List<String> fieldTypeNames = rowSpec.getFieldTypes()
+        .stream()
+        .map(f -> f.getTypeName().getSimple())
+        .collect(Collectors.toList());
+    assertThat(rowSpec.getTypeName().getSimple(), is("ROW"));
+    assertThat(fieldNames, is(Arrays.asList("i", "j")));
+    assertThat(fieldTypeNames, is(Arrays.asList("INTEGER", "INTEGER")));
+  }
+
   private RelDataType struct(RelDataType...relDataTypes) {
     final RelDataTypeFactory.Builder builder = f.typeFactory.builder();
     for (int i = 0; i < relDataTypes.length; i++) {