[UUID 4/8] Server-side predicate evaluation for the logical UUID type

pinot-common/src/main/java/org/apache/pinot/common/request/context/predicate/BaseInPredicate.java

@@ -25,6 +25,7 @@
 import org.apache.pinot.spi.utils.ByteArray;
 import org.apache.pinot.spi.utils.BytesUtils;
 import org.apache.pinot.spi.utils.TimestampUtils;
+import org.apache.pinot.spi.utils.UuidUtils;
 
 
 /// Base predicate for `IN` and `NOT_IN`.
@@ -49,6 +50,7 @@ public abstract class BaseInPredicate extends BasePredicate {
   private volatile int[] _booleanValues;
   private volatile long[] _timestampValues;
   private volatile ByteArray[] _bytesValues;
+  private volatile ByteArray[] _uuidValues;
 
   public BaseInPredicate(ExpressionContext lhs, List<String> values) {
     super(lhs);
@@ -162,4 +164,17 @@ public ByteArray[] getBytesValues() {
     }
     return bigDecimalValues;
   }
+
+  public ByteArray[] getUuidValues() {
+    ByteArray[] uuidValues = _uuidValues;
+    if (uuidValues == null) {
+      int numValues = _values.size();
+      uuidValues = new ByteArray[numValues];
+      for (int i = 0; i < numValues; i++) {
+        uuidValues[i] = new ByteArray(UuidUtils.toBytes(_values.get(i)));
+      }
+      _uuidValues = uuidValues;
+    }
+    return uuidValues;
+  }
 }

pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/EqualsPredicateEvaluatorFactory.java

@@ -31,6 +31,7 @@
 import org.apache.pinot.spi.utils.BooleanUtils;
 import org.apache.pinot.spi.utils.BytesUtils;
 import org.apache.pinot.spi.utils.TimestampUtils;
+import org.apache.pinot.spi.utils.UuidUtils;
 
 
 /// Factory for EQ predicate evaluators.
@@ -76,6 +77,11 @@ public static EqRawPredicateEvaluator newRawValueBasedEvaluator(EqPredicate eqPr
         return new StringRawValueBasedEqPredicateEvaluator(eqPredicate, value);
       case BYTES:
         return new BytesRawValueBasedEqPredicateEvaluator(eqPredicate, BytesUtils.toBytes(value));
+      // UUID is a logical type stored as 16 raw bytes, so -- like TIMESTAMP over LONG above -- convert the literal to
+      // its stored form and reuse the stored-type evaluator. getDataType() then correctly reports the type applySV
+      // consumes (BYTES), per the PredicateEvaluator contract.
+      case UUID:
+        return new BytesRawValueBasedEqPredicateEvaluator(eqPredicate, UuidUtils.toBytes(value));
       default:
         throw new IllegalStateException("Unsupported data type: " + dataType);
     }

pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/InPredicateEvaluatorFactory.java

@@ -144,6 +144,19 @@ public static InRawPredicateEvaluator newRawValueBasedEvaluator(InPredicate inPr
         }
         return new BytesRawValueBasedInPredicateEvaluator(inPredicate, matchingValues);
       }
+      // UUID is a logical type stored as 16 raw bytes, so -- like TIMESTAMP over LONG above -- convert the
+      // literals to their stored form and reuse the stored-type evaluator.
+      case UUID: {
+        ByteArray[] uuidValues = inPredicate.getUuidValues();
+        Set<ByteArray> matchingValues = new ObjectOpenHashSet<>(HashUtil.getMinHashSetSize(uuidValues.length));
+        // NOTE: Add value-by-value to avoid overhead
+        //noinspection ManualArrayToCollectionCopy
+        for (ByteArray value : uuidValues) {
+          //noinspection UseBulkOperation
+          matchingValues.add(value);
+        }
+        return new BytesRawValueBasedInPredicateEvaluator(inPredicate, matchingValues);
+      }
       default:
         throw new IllegalStateException("Unsupported data type: " + dataType);
     }

pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/NotEqualsPredicateEvaluatorFactory.java

@@ -27,6 +27,7 @@
 import org.apache.pinot.spi.utils.BooleanUtils;
 import org.apache.pinot.spi.utils.BytesUtils;
 import org.apache.pinot.spi.utils.TimestampUtils;
+import org.apache.pinot.spi.utils.UuidUtils;
 
 
 /// Factory for NEQ predicate evaluators.
@@ -72,6 +73,11 @@ public static NeqRawPredicateEvaluator newRawValueBasedEvaluator(NotEqPredicate
         return new StringRawValueBasedNeqPredicateEvaluator(notEqPredicate, value);
       case BYTES:
         return new BytesRawValueBasedNeqPredicateEvaluator(notEqPredicate, BytesUtils.toBytes(value));
+      // UUID is a logical type stored as 16 raw bytes, so -- like TIMESTAMP over LONG above -- convert the literal to
+      // its stored form and reuse the stored-type evaluator. getDataType() then correctly reports the type applySV
+      // consumes (BYTES), per the PredicateEvaluator contract.
+      case UUID:
+        return new BytesRawValueBasedNeqPredicateEvaluator(notEqPredicate, UuidUtils.toBytes(value));
       default:
         throw new IllegalStateException("Unsupported data type: " + dataType);
     }

pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/NotInPredicateEvaluatorFactory.java

@@ -144,6 +144,19 @@ public static NotInRawPredicateEvaluator newRawValueBasedEvaluator(NotInPredicat
         }
         return new BytesRawValueBasedNotInPredicateEvaluator(notInPredicate, nonMatchingValues);
       }
+      // UUID is a logical type stored as 16 raw bytes, so -- like TIMESTAMP over LONG above -- convert the
+      // literals to their stored form and reuse the stored-type evaluator.
+      case UUID: {
+        ByteArray[] uuidValues = notInPredicate.getUuidValues();
+        Set<ByteArray> nonMatchingValues = new ObjectOpenHashSet<>(HashUtil.getMinHashSetSize(uuidValues.length));
+        // NOTE: Add value-by-value to avoid overhead
+        //noinspection ManualArrayToCollectionCopy
+        for (ByteArray value : uuidValues) {
+          //noinspection UseBulkOperation
+          nonMatchingValues.add(value);
+        }
+        return new BytesRawValueBasedNotInPredicateEvaluator(notInPredicate, nonMatchingValues);
+      }
       default:
         throw new IllegalStateException("Unsupported data type: " + dataType);
     }

pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/PredicateUtils.java

@@ -32,8 +32,10 @@
 import org.apache.pinot.spi.data.FieldSpec.DataType;
 import org.apache.pinot.spi.utils.BooleanUtils;
 import org.apache.pinot.spi.utils.ByteArray;
+import org.apache.pinot.spi.utils.BytesUtils;
 import org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
 import org.apache.pinot.spi.utils.TimestampUtils;
+import org.apache.pinot.spi.utils.UuidUtils;
 
 
 public class PredicateUtils {
@@ -51,6 +53,15 @@ public static String getStoredValue(String value, DataType dataType) {
         return getStoredBooleanValue(value);
       case TIMESTAMP:
         return getStoredTimestampValue(value);
+      case UUID:
+        // The hex here is a transport encoding for the String-typed lookup APIs, NOT the storage format -- a UUID
+        // column is stored as its raw 16 bytes, and the bytes round-trip unchanged (encode here, decode in the
+        // dictionary). Range bounds reach the dictionary only through Dictionary#insertionIndexOf(String) and
+        // #getDictIdsInRange(String, ...), and the canonical "550e8400-..." form cannot be passed through as-is
+        // because the dashes are not valid hex. Equality and IN avoid this entirely: they resolve UUIDs
+        // byte-natively via Dictionary#indexOf(ByteArray). Adding a matching insertionIndexOf(ByteArray) overload
+        // would let range bounds do the same.
+        return BytesUtils.toHexString(UuidUtils.toBytes(value));
       default:
         return value;
     }
@@ -177,6 +188,15 @@ public static IntSet getDictIdSet(BaseInPredicate inPredicate, Dictionary dictio
           }
         }
         break;
+      case UUID:
+        ByteArray[] uuidValues = inPredicate.getUuidValues();
+        for (ByteArray value : uuidValues) {
+          int dictId = dictionary.indexOf(value);
+          if (dictId >= 0) {
+            dictIdSet.add(dictId);
+          }
+        }
+        break;
       default:
         throw new IllegalStateException("Unsupported data type: " + dataType);
     }

pinot-core/src/main/java/org/apache/pinot/core/operator/filter/predicate/RangePredicateEvaluatorFactory.java

@@ -34,6 +34,7 @@
 import org.apache.pinot.spi.utils.ByteArray;
 import org.apache.pinot.spi.utils.BytesUtils;
 import org.apache.pinot.spi.utils.TimestampUtils;
+import org.apache.pinot.spi.utils.UuidUtils;
 
 
 /// Factory for RANGE predicate evaluators.
@@ -107,6 +108,14 @@ public static RangeRawPredicateEvaluator newRawValueBasedEvaluator(RangePredicat
         return new BytesRawValueBasedRangePredicateEvaluator(rangePredicate,
             lowerUnbounded ? null : BytesUtils.toBytes(lowerBound),
             upperUnbounded ? null : BytesUtils.toBytes(upperBound), lowerInclusive, upperInclusive);
+      // UUID is stored as 16 raw bytes and its unsigned bytewise ordering is exactly UUID ordering, so -- like
+      // TIMESTAMP over LONG above -- convert the bounds to the stored form and reuse the BYTES evaluator. UUID
+      // dictionaries also report getValueType() == BYTES, so the unsorted dictionary-based evaluator dispatches on
+      // BYTES and feeds this the raw 16-byte values, directly comparable to the bounds.
+      case UUID:
+        return new BytesRawValueBasedRangePredicateEvaluator(rangePredicate,
+            lowerUnbounded ? null : UuidUtils.toBytes(lowerBound),
+            upperUnbounded ? null : UuidUtils.toBytes(upperBound), lowerInclusive, upperInclusive);
       default:
         throw new IllegalStateException("Unsupported data type: " + dataType);
     }

pinot-core/src/main/java/org/apache/pinot/core/query/reduce/filter/PredicateRowMatcher.java

@@ -20,11 +20,13 @@
 
 import java.math.BigDecimal;
 import java.sql.Timestamp;
+import java.util.UUID;
 import javax.annotation.Nullable;
 import org.apache.pinot.common.request.context.predicate.Predicate;
 import org.apache.pinot.core.operator.filter.predicate.PredicateEvaluator;
 import org.apache.pinot.core.operator.filter.predicate.PredicateEvaluatorProvider;
 import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.UuidUtils;
 
 
 /// Predicate matcher.
@@ -78,6 +80,8 @@ public boolean isMatch(Object[] row) {
         return _predicateEvaluator.applySV((String) value);
       case BYTES:
         return _predicateEvaluator.applySV((byte[]) value);
+      case UUID:
+        return _predicateEvaluator.applySV(UuidUtils.toBytes((UUID) value));
       default:
         throw new IllegalStateException();
     }

pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/NoDictionaryEqualsPredicateEvaluatorsTest.java

@@ -19,14 +19,18 @@
 package org.apache.pinot.core.operator.filter.predicate;
 
 import java.math.BigDecimal;
+import java.util.Arrays;
+import java.util.Locale;
 import java.util.Random;
+import java.util.UUID;
 import org.apache.commons.lang3.ArrayUtils;
 import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.pinot.common.request.context.ExpressionContext;
 import org.apache.pinot.common.request.context.predicate.EqPredicate;
 import org.apache.pinot.common.request.context.predicate.NotEqPredicate;
 import org.apache.pinot.spi.data.FieldSpec;
 import org.apache.pinot.spi.utils.BytesUtils;
+import org.apache.pinot.spi.utils.UuidUtils;
 import org.testng.Assert;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
@@ -322,4 +326,43 @@ public void testBytesPredicateEvaluators() {
           !ArrayUtils.contains(randomBytesArray, stringValue));
     }
   }
+
+  @Test
+  public void testUuidPredicateEvaluators() {
+    UUID uuidValue = UUID.fromString("550e8400-e29b-41d4-a716-446655440000");
+    byte[] uuidBytes = UuidUtils.toBytes(uuidValue);
+    // Predicate literals reach the evaluator as UUID strings. Use an upper-cased one to pin down that the hex digits
+    // are matched case-insensitively rather than compared as raw strings.
+    String stringValue = uuidValue.toString().toUpperCase(Locale.ROOT);
+
+    EqPredicate eqPredicate = new EqPredicate(COLUMN_EXPRESSION, stringValue);
+    PredicateEvaluator eqPredicateEvaluator =
+        EqualsPredicateEvaluatorFactory.newRawValueBasedEvaluator(eqPredicate, FieldSpec.DataType.UUID);
+
+    NotEqPredicate notEqPredicate = new NotEqPredicate(COLUMN_EXPRESSION, stringValue);
+    PredicateEvaluator neqPredicateEvaluator =
+        NotEqualsPredicateEvaluatorFactory.newRawValueBasedEvaluator(notEqPredicate, FieldSpec.DataType.UUID);
+
+    // getDataType() reports the type applySV consumes, not the column's logical type -- exactly as a TIMESTAMP
+    // column's evaluator reports LONG. UUID literals are converted to their 16-byte stored form up front, so the
+    // BYTES raw evaluator is reused as-is and reports BYTES.
+    Assert.assertEquals(eqPredicateEvaluator.getDataType(), FieldSpec.DataType.BYTES);
+    Assert.assertEquals(neqPredicateEvaluator.getDataType(), FieldSpec.DataType.BYTES);
+
+    Assert.assertTrue(eqPredicateEvaluator.applySV(uuidBytes));
+    Assert.assertFalse(neqPredicateEvaluator.applySV(uuidBytes));
+
+    // A UUID differing only in the last byte must not match, guarding against a truncated comparison.
+    byte[] nearMissBytes = Arrays.copyOf(uuidBytes, uuidBytes.length);
+    nearMissBytes[nearMissBytes.length - 1] ^= 0x01;
+    Assert.assertFalse(eqPredicateEvaluator.applySV(nearMissBytes));
+    Assert.assertTrue(neqPredicateEvaluator.applySV(nearMissBytes));
+
+    for (int i = 0; i < 100; i++) {
+      byte[] randomUuidBytes = UuidUtils.toBytes(UUID.randomUUID());
+      boolean matches = Arrays.equals(randomUuidBytes, uuidBytes);
+      Assert.assertEquals(eqPredicateEvaluator.applySV(randomUuidBytes), matches);
+      Assert.assertEquals(neqPredicateEvaluator.applySV(randomUuidBytes), !matches);
+    }
+  }
 }

pinot-core/src/test/java/org/apache/pinot/core/operator/filter/predicate/NoDictionaryInPredicateEvaluatorTest.java

@@ -39,6 +39,7 @@
 import org.apache.pinot.common.request.context.predicate.NotInPredicate;
 import org.apache.pinot.spi.data.FieldSpec;
 import org.apache.pinot.spi.utils.BytesUtils;
+import org.apache.pinot.spi.utils.UuidUtils;
 import org.testng.Assert;
 import org.testng.annotations.BeforeClass;
 import org.testng.annotations.Test;
@@ -363,4 +364,76 @@ public void testBytesPredicateEvaluators() {
     Assert.assertTrue(inPredicateEvaluator.applyMV(multiValues, NUM_MULTI_VALUES));
     Assert.assertFalse(notInPredicateEvaluator.applyMV(multiValues, NUM_MULTI_VALUES));
   }
+
+  @Test
+  public void testUuidPredicateEvaluators() {
+    List<String> uuidStrings = new ArrayList<>(NUM_PREDICATE_VALUES);
+    Set<String> uuidStringSet = new HashSet<>();
+
+    for (int i = 0; i < NUM_PREDICATE_VALUES; i++) {
+      String uuidString = java.util.UUID.randomUUID().toString();
+      uuidStrings.add(uuidString);
+      uuidStringSet.add(uuidString);
+    }
+
+    InPredicate inPredicate = new InPredicate(COLUMN_EXPRESSION, uuidStrings);
+    PredicateEvaluator inPredicateEvaluator =
+        InPredicateEvaluatorFactory.newRawValueBasedEvaluator(inPredicate, FieldSpec.DataType.UUID);
+
+    NotInPredicate notInPredicate = new NotInPredicate(COLUMN_EXPRESSION, uuidStrings);
+    PredicateEvaluator notInPredicateEvaluator =
+        NotInPredicateEvaluatorFactory.newRawValueBasedEvaluator(notInPredicate, FieldSpec.DataType.UUID);
+
+    // getDataType() reports the type applySV consumes, not the column's logical type -- exactly as a TIMESTAMP
+    // column's evaluator reports LONG. UUID literals are converted to their 16-byte stored form up front, so the
+    // BYTES raw evaluator is reused as-is and reports BYTES.
+    Assert.assertEquals(inPredicateEvaluator.getDataType(), FieldSpec.DataType.BYTES);
+    Assert.assertEquals(notInPredicateEvaluator.getDataType(), FieldSpec.DataType.BYTES);
+
+    for (String uuidString : uuidStringSet) {
+      byte[] uuidBytes = UuidUtils.toBytes(uuidString);
+      Assert.assertTrue(inPredicateEvaluator.applySV(uuidBytes));
+      Assert.assertFalse(notInPredicateEvaluator.applySV(uuidBytes));
+    }
+
+    for (int i = 0; i < NUM_PREDICATE_VALUES; i++) {
+      byte[] value = UuidUtils.toBytes(java.util.UUID.randomUUID());
+      boolean expected = uuidStringSet.contains(UuidUtils.toString(value));
+      Assert.assertEquals(inPredicateEvaluator.applySV(value), expected);
+      Assert.assertEquals(notInPredicateEvaluator.applySV(value), !expected);
+    }
+  }
+
+  /// The BYTES/UUID raw evaluators key their matching set on the raw `byte[]` so that `applySV` does not
+  /// wrap every scanned value. That only works if the set compares by *content*: with identity semantics a
+  /// scanned array would never match a predicate array, and IN would silently return nothing while NOT IN returned
+  /// everything. Probe with arrays that are equal but deliberately not the same instance.
+  @Test
+  public void testBytesAndUuidPredicatesMatchByValueNotIdentity() {
+    String uuidString = "550e8400-e29b-41d4-a716-446655440000";
+    String bytesHex = "0a1b2c3d";
+
+    for (Object[] testCase : new Object[][]{
+        {FieldSpec.DataType.UUID, uuidString, UuidUtils.toBytes(uuidString)},
+        {FieldSpec.DataType.BYTES, bytesHex, BytesUtils.toBytes(bytesHex)}
+    }) {
+      FieldSpec.DataType dataType = (FieldSpec.DataType) testCase[0];
+      List<String> values = List.of((String) testCase[1]);
+      byte[] probe = ((byte[]) testCase[2]).clone();
+
+      PredicateEvaluator inEvaluator = InPredicateEvaluatorFactory.newRawValueBasedEvaluator(
+          new InPredicate(COLUMN_EXPRESSION, values), dataType);
+      PredicateEvaluator notInEvaluator = NotInPredicateEvaluatorFactory.newRawValueBasedEvaluator(
+          new NotInPredicate(COLUMN_EXPRESSION, values), dataType);
+
+      Assert.assertTrue(inEvaluator.applySV(probe), dataType + " IN must match an equal-but-distinct array");
+      Assert.assertFalse(notInEvaluator.applySV(probe),
+          dataType + " NOT IN must not match an equal-but-distinct array");
+
+      byte[] different = probe.clone();
+      different[0] ^= 0xff;
+      Assert.assertFalse(inEvaluator.applySV(different), dataType + " IN must not match a different value");
+      Assert.assertTrue(notInEvaluator.applySV(different), dataType + " NOT IN must match a different value");
+    }
+  }
 }