perf(strings): SIMD-vectorize lengthUnicode

velox/functions/lib/string/StringCore.h

@@ -264,22 +264,38 @@ applyAppendersRecursive(TOutStr& output, Func appenderFunc, Funcs... funcs) {
  */
 FOLLY_ALWAYS_INLINE int64_t
 lengthUnicode(const char* inputBuffer, size_t bufferLength) {
-  // First address after the last byte in the buffer
-  auto buffEndAddress = inputBuffer + bufferLength;
-  auto currentChar = inputBuffer;
-  int64_t size = 0;
-  while (currentChar < buffEndAddress) {
-    // This function detects bytes that come after the first byte in a
-    // multi-byte UTF-8 character (provided that the string is valid UTF-8). We
-    // increment size only for the first byte so that we treat all bytes as part
-    // of a single character.
-    if (!utf_cont(*currentChar)) {
-      size++;
+  // Character count equals the number of non-continuation bytes, i.e. bytes
+  // where (b & 0xC0) != 0x80. SIMD-count per block; scalar tail for remainder.
+  constexpr size_t kBatchSize = xsimd::batch<uint8_t>::size;
+  const auto* input = reinterpret_cast<const uint8_t*>(inputBuffer);
+  size_t position = 0;
+  int64_t numChars = 0;
+
+  if (bufferLength >= kBatchSize) {
+    const auto highBitMask = xsimd::broadcast<uint8_t>(0x80);
+    const auto continuationMask = xsimd::broadcast<uint8_t>(0xc0);
+
+    while (position + kBatchSize <= bufferLength) {
+      auto batch = xsimd::batch<uint8_t>::load_unaligned(input + position);
+      if (LIKELY(!xsimd::any(batch >= highBitMask))) {
+        numChars += kBatchSize; // Pure ASCII block.
+      } else {
+        const auto isCharStart = (batch & continuationMask) != highBitMask;
+        const uint64_t mask =
+            static_cast<uint64_t>(simd::toBitMask(isCharStart));
+        numChars += __builtin_popcountll(mask);
+      }
+      position += kBatchSize;
     }
+  }
 
-    currentChar++;
+  while (position < bufferLength) {
+    if (!utf_cont(static_cast<char>(input[position]))) {
+      numChars++;
+    }
+    position++;
   }
-  return size;
+  return numChars;
 }
 
 namespace detail {

velox/functions/prestosql/benchmarks/StringAsciiUTFFunctionBenchmarks.cpp

@@ -58,6 +58,31 @@ class StringAsciiUTFFunctionBenchmark
     doRun(exprSet, rowVector);
   }
 
+  void runLength(bool utf, size_t stringLength) {
+    folly::BenchmarkSuspender suspender;
+
+    VectorFuzzer::Options opts;
+    if (utf) {
+      opts.charEncodings.clear();
+      opts.charEncodings = {
+          UTF8CharList::UNICODE_CASE_SENSITIVE,
+          UTF8CharList::EXTENDED_UNICODE,
+          UTF8CharList::MATHEMATICAL_SYMBOLS};
+    }
+
+    opts.stringLength = stringLength;
+    opts.vectorSize = 100'000;
+    VectorFuzzer fuzzer(opts, execCtx_.pool());
+    auto vector = fuzzer.fuzzFlat(VARCHAR());
+
+    auto rowVector = vectorMaker_.rowVector({vector});
+    auto exprSet = compileExpression("length(c0)", rowVector->type());
+
+    suspender.dismiss();
+
+    doRun(exprSet, rowVector);
+  }
+
   void runSubStr(bool utf) {
     folly::BenchmarkSuspender suspender;
 
@@ -126,6 +151,26 @@ class StringAsciiUTFFunctionBenchmark
   }
 };
 
+BENCHMARK(utfLength_100) {
+  StringAsciiUTFFunctionBenchmark benchmark;
+  benchmark.runLength(true, 100);
+}
+
+BENCHMARK_RELATIVE(asciiLength_100) {
+  StringAsciiUTFFunctionBenchmark benchmark;
+  benchmark.runLength(false, 100);
+}
+
+BENCHMARK(utfLength_1024) {
+  StringAsciiUTFFunctionBenchmark benchmark;
+  benchmark.runLength(true, 1024);
+}
+
+BENCHMARK_RELATIVE(asciiLength_1024) {
+  StringAsciiUTFFunctionBenchmark benchmark;
+  benchmark.runLength(false, 1024);
+}
+
 BENCHMARK(utfLower) {
   StringAsciiUTFFunctionBenchmark benchmark;
   benchmark.runUpperLower("lower", true);
@@ -189,6 +234,8 @@ BENCHMARK_RELATIVE(aciiRPad) {
 //============================================================================
 int main(int argc, char** argv) {
   folly::Init init{&argc, &argv};
+  facebook::velox::memory::MemoryManager::initialize(
+      facebook::velox::memory::MemoryManager::Options{});
 
   folly::runBenchmarks();
   return 0;