[Do not merge] URN Data Type

.gitignore

@@ -58,3 +58,4 @@ kubernetes/helm/**/Chart.lock
 #Develocity
 .mvn/.gradle-enterprise/
 .mvn/.develocity/
+.claude/

pinot-common/src/main/java/org/apache/pinot/common/datablock/DataBlockEquals.java

@@ -295,6 +295,7 @@ public boolean equals(DataBlock left, DataBlock right) {
               break;
             case STRING:
             case JSON:
+            case URN:
               for (int did = 0; did < numRows; did++) {
                 if (!left.getString(did, colId).equals(right.getString(did, colId))) {
                   if (_failOnFalse) {

pinot-common/src/main/java/org/apache/pinot/common/function/FunctionUtils.java

@@ -193,6 +193,7 @@ public static RelDataType getRelDataType(RelDataTypeFactory typeFactory, Class<?
         return typeFactory.createSqlType(SqlTypeName.TIMESTAMP);
       case STRING:
       case JSON:
+      case URN:
         return typeFactory.createSqlType(SqlTypeName.VARCHAR);
       case BYTES:
         return typeFactory.createSqlType(SqlTypeName.VARBINARY);

pinot-common/src/main/java/org/apache/pinot/common/function/scalar/UrnFunctions.java

@@ -0,0 +1,190 @@
+/**
+ * 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.pinot.common.function.scalar;
+
+import javax.annotation.Nullable;
+import org.apache.pinot.spi.annotations.ScalarFunction;
+import org.apache.pinot.spi.utils.Urn;
+
+
+/**
+ * Scalar functions for working with URN-typed columns.
+ *
+ * <p>All functions accept a URN string and return a component of it. They are null-safe: a
+ * {@code null} input always produces a {@code null} output. An unparseable (malformed) URN
+ * produces a {@code null} rather than throwing so that queries over mixed-quality data do not fail
+ * entirely.
+ *
+ * <p>Example usage in SQL:
+ * <pre>
+ *   SELECT urnEntity(memberId), COUNT(*) FROM myTable GROUP BY 1
+ *   SELECT urnKey(memberId) FROM myTable WHERE urnEntity(memberId) = 'urn:li:memberId'
+ *   SELECT * FROM myTable WHERE urnEntityType(memberId) = 'memberId'
+ *   SELECT urnIsComposite(datasetId), urnKeyPart(datasetId, 1) FROM myTable
+ * </pre>
+ */
+public class UrnFunctions {
+  private UrnFunctions() {
+  }
+
+  /**
+   * Returns the fully-qualified entity type prefix of a URN, e.g. {@code "urn:li:memberId"} for
+   * {@code "urn:li:memberId:123456"}.
+   *
+   * <p>This is the canonical string to use when grouping or filtering on entity type — it includes
+   * the full namespace, so {@code "urn:li:memberId"} is unambiguous across different namespaces.
+   */
+  @Nullable
+  @ScalarFunction(names = {"urnEntity", "urn_entity"}, nullableParameters = true)
+  public static String urnEntity(@Nullable String urnStr) {
+    if (urnStr == null) {
+      return null;
+    }
+    Urn urn = Urn.tryParse(urnStr);
+    return urn != null ? urn.entityPrefix() : null;
+  }
+
+  /**
+   * Returns the entity namespace component, e.g. {@code "li"} for {@code "urn:li:memberId:123"}.
+   */
+  @Nullable
+  @ScalarFunction(names = {"urnNamespace", "urn_namespace"}, nullableParameters = true)
+  public static String urnNamespace(@Nullable String urnStr) {
+    if (urnStr == null) {
+      return null;
+    }
+    Urn urn = Urn.tryParse(urnStr);
+    return urn != null ? urn.namespace() : null;
+  }
+
+  /**
+   * Returns the entity type component (without namespace), e.g. {@code "memberId"} for
+   * {@code "urn:li:memberId:123456"}.
+   *
+   * <p>Useful for grouping by entity type across namespaces or when the namespace is constant.
+   */
+  @Nullable
+  @ScalarFunction(names = {"urnEntityType", "urn_entity_type"}, nullableParameters = true)
+  public static String urnEntityType(@Nullable String urnStr) {
+    if (urnStr == null) {
+      return null;
+    }
+    Urn urn = Urn.tryParse(urnStr);
+    return urn != null ? urn.entityType() : null;
+  }
+
+  /**
+   * Returns the key component of a simple (non-composite) URN as a string, e.g. {@code "123456"}
+   * for {@code "urn:li:memberId:123456"}.
+   *
+   * <p>For composite URNs the full tuple string including parentheses is returned, e.g.
+   * {@code "(urn:li:dataPlatform:pinot,memberEducation,EI)"} for a dataset URN.
+   */
+  @Nullable
+  @ScalarFunction(names = {"urnKey", "urn_key"}, nullableParameters = true)
+  public static String urnKey(@Nullable String urnStr) {
+    if (urnStr == null) {
+      return null;
+    }
+    Urn urn = Urn.tryParse(urnStr);
+    if (urn == null) {
+      return null;
+    }
+    if (urn.isSimple()) {
+      return urn.simpleKey();
+    }
+    // Composite: reconstruct the key tuple string
+    StringBuilder sb = new StringBuilder("(");
+    for (int i = 0; i < urn.parts().size(); i++) {
+      if (i > 0) {
+        sb.append(',');
+      }
+      sb.append(urn.parts().get(i).toString());
+    }
+    sb.append(')');
+    return sb.toString();
+  }
+
+  /**
+   * Returns the key of a simple URN as a long, or {@code Long.MIN_VALUE} if the key is not
+   * numeric. Returns {@code null} on null input or unparseable URN.
+   *
+   * <p>This is the most efficient function for numeric-key URN columns — it allows the query
+   * engine to operate directly on long values without repeated string-to-long conversion.
+   */
+  @Nullable
+  @ScalarFunction(names = {"urnKeyLong", "urn_key_long"}, nullableParameters = true)
+  public static Long urnKeyLong(@Nullable String urnStr) {
+    if (urnStr == null) {
+      return null;
+    }
+    Urn urn = Urn.tryParse(urnStr);
+    if (urn == null || !urn.isSimple()) {
+      return null;
+    }
+    try {
+      return Long.parseLong(urn.simpleKey());
+    } catch (NumberFormatException e) {
+      return null;
+    }
+  }
+
+  /**
+   * Returns {@code true} if the URN is composite (its key is a tuple), {@code false} if it is
+   * simple. Returns {@code null} on null input or unparseable URN.
+   */
+  @Nullable
+  @ScalarFunction(names = {"urnIsComposite", "urn_is_composite"}, nullableParameters = true)
+  public static Boolean urnIsComposite(@Nullable String urnStr) {
+    if (urnStr == null) {
+      return null;
+    }
+    Urn urn = Urn.tryParse(urnStr);
+    return urn != null ? !urn.isSimple() : null;
+  }
+
+  /**
+   * Returns the {@code index}-th part of a composite URN's tuple key as a string (0-based).
+   * Returns {@code null} for simple URNs, out-of-range indices, null input, or unparseable URN.
+   *
+   * <p>Example: {@code urnKeyPart('urn:li:dataset:(urn:li:dataPlatform:pinot,foo,PROD)', 1)}
+   * returns {@code "foo"}.
+   */
+  @Nullable
+  @ScalarFunction(names = {"urnKeyPart", "urn_key_part"}, nullableParameters = true)
+  public static String urnKeyPart(@Nullable String urnStr, int index) {
+    if (urnStr == null) {
+      return null;
+    }
+    Urn urn = Urn.tryParse(urnStr);
+    if (urn == null || urn.isSimple() || index < 0 || index >= urn.parts().size()) {
+      return null;
+    }
+    return urn.parts().get(index).toString();
+  }
+
+  /**
+   * Returns {@code true} if the string is a valid, parseable URN; {@code false} otherwise.
+   * Never returns {@code null}; a null input returns {@code false}.
+   */
+  @ScalarFunction(names = {"isUrn", "is_urn"}, nullableParameters = true)
+  public static boolean isUrn(@Nullable String value) {
+    return Urn.isUrn(value);
+  }
+}

pinot-common/src/main/java/org/apache/pinot/common/response/encoder/ArrowResponseEncoder.java

@@ -108,6 +108,7 @@ private VectorSchemaRoot createVectorSchemaRoot(ResultTable resultTable, DataSch
         case BYTES:
         case BIG_DECIMAL:
         case JSON:
+        case URN:
         case OBJECT:
           field = new Field(colName, FieldType.nullable(new ArrowType.Utf8()), null);
           vector = new VarCharVector(colName, ALLOCATOR);
@@ -234,6 +235,7 @@ private VectorSchemaRoot createVectorSchemaRoot(ResultTable resultTable, DataSch
             case BYTES:
             case BIG_DECIMAL:
             case JSON:
+            case URN:
             case OBJECT:
               byte[] bytes = ((String) value).getBytes(StandardCharsets.UTF_8);
               ((VarCharVector) vector).setSafe(rowIndex, bytes);
@@ -410,6 +412,7 @@ public ResultTable decodeResultTable(byte[] bytes, int rowSize, DataSchema schem
             case BYTES:
             case BIG_DECIMAL:
             case JSON:
+            case URN:
             case OBJECT:
               row[col] = new String(((VarCharVector) vector).get(i), StandardCharsets.UTF_8);
               break;

pinot-common/src/main/java/org/apache/pinot/common/response/encoder/JsonResponseEncoder.java

@@ -223,6 +223,7 @@ private static Object extractValue(DataSchema.ColumnDataType columnDataType, Jso
       case BYTES:
       case TIMESTAMP:
       case JSON:
+      case URN:
       case BIG_DECIMAL:
       case OBJECT:
         return jsonValue.textValue();

pinot-common/src/main/java/org/apache/pinot/common/utils/DataSchema.java

@@ -290,6 +290,12 @@ public RelDataType toType(RelDataTypeFactory typeFactory) {
         return typeFactory.createSqlType(SqlTypeName.VARCHAR);
       }
     },
+    URN(STRING, NullValuePlaceHolder.STRING) {
+      @Override
+      public RelDataType toType(RelDataTypeFactory typeFactory) {
+        return typeFactory.createSqlType(SqlTypeName.VARCHAR);
+      }
+    },
     MAP(NullValuePlaceHolder.MAP) {
       @Override
       public RelDataType toType(RelDataTypeFactory typeFactory) {
@@ -452,6 +458,8 @@ public DataType toDataType() {
           return DataType.STRING;
         case JSON:
           return DataType.JSON;
+        case URN:
+          return DataType.URN;
         case BYTES:
         case BYTES_ARRAY:
           return DataType.BYTES;
@@ -571,6 +579,7 @@ public Serializable convert(Object value) {
           return new Timestamp((long) value);
         case STRING:
         case JSON:
+        case URN:
           return value.toString();
         case BYTES:
           return ((ByteArray) value).getBytes();
@@ -641,6 +650,7 @@ public Serializable convertAndFormat(Object value) {
           return new Timestamp((long) value).toString();
         case STRING:
         case JSON:
+        case URN:
           return value.toString();
         case BYTES:
           return ((ByteArray) value).toHexString();
@@ -844,6 +854,8 @@ public static ColumnDataType fromDataTypeSV(DataType dataType) {
           return STRING;
         case JSON:
           return JSON;
+        case URN:
+          return URN;
         case BYTES:
           return BYTES;
         case MAP:

pinot-common/src/main/java/org/apache/pinot/common/utils/PinotDataType.java

@@ -1493,6 +1493,11 @@ public static PinotDataType getPinotDataTypeForIngestion(FieldSpec fieldSpec) {
           return JSON;
         }
         throw new IllegalStateException("There is no multi-value type for JSON");
+      case URN:
+        if (fieldSpec.isSingleValueField()) {
+          return STRING;
+        }
+        throw new IllegalStateException("There is no multi-value type for URN");
       case STRING:
         return fieldSpec.isSingleValueField() ? STRING : STRING_ARRAY;
       case BYTES:
@@ -1532,6 +1537,8 @@ public static PinotDataType getPinotDataTypeForExecution(ColumnDataType columnDa
         return STRING;
       case JSON:
         return JSON;
+      case URN:
+        return STRING;
       case BYTES:
         return BYTES;
       case OBJECT: