[Feature](function) Support function DAMERAU_LEVENSHTEIN_DISTANCE (#65278)

### Release note

Support function `DAMERAU_LEVENSHTEIN_DISTANCE`

doc: https://github.com/apache/doris-website/pull/3971

be/benchmark/benchmark_damerau_levenshtein.hpp

@@ -0,0 +1,245 @@
+// 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.
+
+#include <benchmark/benchmark.h>
+
+#include <algorithm>
+#include <cstdint>
+#include <string>
+#include <string_view>
+#include <unordered_map>
+#include <vector>
+
+#include "core/string_ref.h"
+#include "util/simd/vstring_function.h"
+
+namespace doris {
+
+namespace {
+
+using BenchmarkSymbolId = int32_t;
+
+std::string make_utf8_string(size_t chars, size_t seed, bool transpose_adjacent) {
+    static const std::vector<std::string> symbols = {"数", "据", "库", "查", "询", "表",
+                                                     "列", "你", "好", "测", "试", "节",
+                                                     "点", "分", "区", "行"};
+
+    std::string result;
+    result.reserve(chars * 3);
+    for (size_t i = 0; i < chars; ++i) {
+        size_t symbol_index = i;
+        if (transpose_adjacent && chars > 2) {
+            if (i == 1) {
+                symbol_index = 2;
+            } else if (i == 2) {
+                symbol_index = 1;
+            }
+        }
+        result += symbols[(symbol_index * 7 + seed) % symbols.size()];
+    }
+    return result;
+}
+
+void append_utf8_symbols(std::string_view str, const std::vector<size_t>& offsets,
+                         std::vector<BenchmarkSymbolId>& symbols,
+                         std::unordered_map<std::string_view, BenchmarkSymbolId>& symbol_ids,
+                         BenchmarkSymbolId& next_symbol) {
+    symbols.reserve(symbols.size() + offsets.size());
+    for (size_t i = 0; i < offsets.size(); ++i) {
+        const size_t offset = offsets[i];
+        const size_t next_offset = i + 1 < offsets.size() ? offsets[i + 1] : str.size();
+        const std::string_view symbol(str.data() + offset, next_offset - offset);
+        auto [it, inserted] = symbol_ids.emplace(symbol, next_symbol);
+        if (inserted) {
+            ++next_symbol;
+        }
+        symbols.push_back(it->second);
+    }
+}
+
+int32_t by_symbols(const std::vector<BenchmarkSymbolId>& left,
+                   const std::vector<BenchmarkSymbolId>& right, size_t alphabet_size) {
+    const size_t m = left.size();
+    const size_t n = right.size();
+    if (m == 0) {
+        return static_cast<int32_t>(n);
+    }
+    if (n == 0) {
+        return static_cast<int32_t>(m);
+    }
+
+    const int32_t max_distance = static_cast<int32_t>(m + n);
+    std::vector<int32_t> last_row(alphabet_size, 0);
+    std::vector<int32_t> dist((m + 2) * (n + 2), 0);
+    auto at = [&](size_t i, size_t j) -> int32_t& { return dist[i * (n + 2) + j]; };
+
+    at(0, 0) = max_distance;
+    for (size_t i = 0; i <= m; ++i) {
+        at(i + 1, 0) = max_distance;
+        at(i + 1, 1) = static_cast<int32_t>(i);
+    }
+    for (size_t j = 0; j <= n; ++j) {
+        at(0, j + 1) = max_distance;
+        at(1, j + 1) = static_cast<int32_t>(j);
+    }
+
+    for (size_t i = 1; i <= m; ++i) {
+        int32_t last_match_col = 0;
+        for (size_t j = 1; j <= n; ++j) {
+            const int32_t last_match_row = last_row[right[j - 1]];
+            const int32_t transposition_col = last_match_col;
+            int32_t cost = 1;
+            if (left[i - 1] == right[j - 1]) {
+                cost = 0;
+                last_match_col = static_cast<int32_t>(j);
+            }
+
+            const int32_t replace_cost = at(i, j) + cost;
+            const int32_t insert_cost = at(i + 1, j) + 1;
+            const int32_t delete_cost = at(i, j + 1) + 1;
+            const int32_t transpose_cost = at(last_match_row, transposition_col) +
+                                           static_cast<int32_t>(i - last_match_row - 1) + 1 +
+                                           static_cast<int32_t>(j - transposition_col - 1);
+            at(i + 1, j + 1) = std::min(std::min(replace_cost, insert_cost),
+                                        std::min(delete_cost, transpose_cost));
+        }
+        last_row[left[i - 1]] = static_cast<int32_t>(i);
+    }
+
+    return at(m + 1, n + 1);
+}
+
+int32_t damerau_utf8_symbol_mapped(std::string_view left, const std::vector<size_t>& left_offsets,
+                                   std::string_view right,
+                                   const std::vector<size_t>& right_offsets) {
+    std::vector<BenchmarkSymbolId> left_symbols;
+    std::vector<BenchmarkSymbolId> right_symbols;
+    std::unordered_map<std::string_view, BenchmarkSymbolId> symbol_ids;
+    symbol_ids.reserve(left_offsets.size() + right_offsets.size());
+    BenchmarkSymbolId next_symbol = 0;
+    append_utf8_symbols(left, left_offsets, left_symbols, symbol_ids, next_symbol);
+    append_utf8_symbols(right, right_offsets, right_symbols, symbol_ids, next_symbol);
+    return by_symbols(left_symbols, right_symbols, next_symbol);
+}
+
+std::string_view utf8_symbol_at(std::string_view str, const std::vector<size_t>& offsets,
+                                size_t index) {
+    const size_t offset = offsets[index];
+    const size_t next_offset = index + 1 < offsets.size() ? offsets[index + 1] : str.size();
+    return {str.data() + offset, next_offset - offset};
+}
+
+int32_t damerau_utf8_direct_compare(std::string_view left, const std::vector<size_t>& left_offsets,
+                                    std::string_view right,
+                                    const std::vector<size_t>& right_offsets) {
+    const size_t m = left_offsets.size();
+    const size_t n = right_offsets.size();
+    if (m == 0) {
+        return static_cast<int32_t>(n);
+    }
+    if (n == 0) {
+        return static_cast<int32_t>(m);
+    }
+
+    const int32_t max_distance = static_cast<int32_t>(m + n);
+    std::unordered_map<std::string_view, int32_t> last_row;
+    last_row.reserve(m + n);
+    std::vector<int32_t> dist((m + 2) * (n + 2), 0);
+    auto at = [&](size_t i, size_t j) -> int32_t& { return dist[i * (n + 2) + j]; };
+
+    at(0, 0) = max_distance;
+    for (size_t i = 0; i <= m; ++i) {
+        at(i + 1, 0) = max_distance;
+        at(i + 1, 1) = static_cast<int32_t>(i);
+    }
+    for (size_t j = 0; j <= n; ++j) {
+        at(0, j + 1) = max_distance;
+        at(1, j + 1) = static_cast<int32_t>(j);
+    }
+
+    for (size_t i = 1; i <= m; ++i) {
+        int32_t last_match_col = 0;
+        const std::string_view left_symbol = utf8_symbol_at(left, left_offsets, i - 1);
+        for (size_t j = 1; j <= n; ++j) {
+            const std::string_view right_symbol = utf8_symbol_at(right, right_offsets, j - 1);
+            auto it = last_row.find(right_symbol);
+            const int32_t last_match_row = it == last_row.end() ? 0 : it->second;
+            const int32_t transposition_col = last_match_col;
+            int32_t cost = 1;
+            if (left_symbol == right_symbol) {
+                cost = 0;
+                last_match_col = static_cast<int32_t>(j);
+            }
+
+            const int32_t replace_cost = at(i, j) + cost;
+            const int32_t insert_cost = at(i + 1, j) + 1;
+            const int32_t delete_cost = at(i, j + 1) + 1;
+            const int32_t transpose_cost = at(last_match_row, transposition_col) +
+                                           static_cast<int32_t>(i - last_match_row - 1) + 1 +
+                                           static_cast<int32_t>(j - transposition_col - 1);
+            at(i + 1, j + 1) = std::min(std::min(replace_cost, insert_cost),
+                                        std::min(delete_cost, transpose_cost));
+        }
+        last_row[left_symbol] = static_cast<int32_t>(i);
+    }
+
+    return at(m + 1, n + 1);
+}
+
+void get_utf8_offsets(const std::string& str, std::vector<size_t>& offsets) {
+    const StringRef ref(str.data(), str.size());
+    simd::VStringFunctions::get_utf8_char_offsets(ref, offsets);
+}
+
+void BM_DamerauLevenshtein_UTF8(benchmark::State& state, bool use_symbol_mapped, size_t chars) {
+    const std::string left = make_utf8_string(chars, 0, false);
+    const std::string right = make_utf8_string(chars, 0, true);
+
+    std::vector<size_t> left_offsets;
+    std::vector<size_t> right_offsets;
+    get_utf8_offsets(left, left_offsets);
+    get_utf8_offsets(right, right_offsets);
+
+    for (auto _ : state) {
+        auto distance =
+                use_symbol_mapped
+                        ? damerau_utf8_symbol_mapped(left, left_offsets, right, right_offsets)
+                        : damerau_utf8_direct_compare(left, left_offsets, right, right_offsets);
+        benchmark::DoNotOptimize(distance);
+    }
+}
+
+} // namespace
+
+BENCHMARK_CAPTURE(BM_DamerauLevenshtein_UTF8, SymbolMapped_16, true, 16)
+        ->Unit(benchmark::kMicrosecond);
+BENCHMARK_CAPTURE(BM_DamerauLevenshtein_UTF8, DirectCompare_16, false, 16)
+        ->Unit(benchmark::kMicrosecond);
+BENCHMARK_CAPTURE(BM_DamerauLevenshtein_UTF8, SymbolMapped_64, true, 64)
+        ->Unit(benchmark::kMicrosecond);
+BENCHMARK_CAPTURE(BM_DamerauLevenshtein_UTF8, DirectCompare_64, false, 64)
+        ->Unit(benchmark::kMicrosecond);
+BENCHMARK_CAPTURE(BM_DamerauLevenshtein_UTF8, SymbolMapped_128, true, 128)
+        ->Unit(benchmark::kMicrosecond);
+BENCHMARK_CAPTURE(BM_DamerauLevenshtein_UTF8, DirectCompare_128, false, 128)
+        ->Unit(benchmark::kMicrosecond);
+BENCHMARK_CAPTURE(BM_DamerauLevenshtein_UTF8, SymbolMapped_256, true, 256)
+        ->Unit(benchmark::kMicrosecond);
+BENCHMARK_CAPTURE(BM_DamerauLevenshtein_UTF8, DirectCompare_256, false, 256)
+        ->Unit(benchmark::kMicrosecond);
+
+} // namespace doris

be/benchmark/benchmark_main.cpp

@@ -23,6 +23,7 @@
 #include "benchmark_column_array_view.hpp"
 #include "benchmark_column_array_view_distance.hpp"
 #include "benchmark_column_view.hpp"
+#include "benchmark_damerau_levenshtein.hpp"
 #include "benchmark_fastunion.hpp"
 #include "benchmark_fmod.hpp"
 #include "benchmark_hll_merge.hpp"

be/src/exprs/function/function_levenshtein.cpp

@@ -16,10 +16,16 @@
 // under the License.
 
 #include <algorithm>
+#include <array>
+#include <limits>
+#include <string_view>
+#include <unordered_map>
 #include <vector>
 
 #include "common/status.h"
+#include "core/custom_allocator.h"
 #include "core/data_type/data_type_number.h"
+#include "core/pod_array.h"
 #include "core/string_ref.h"
 #include "exprs/function/function_totype.h"
 #include "exprs/function/simple_function_factory.h"
@@ -31,81 +37,74 @@ struct NameLevenshtein {
     static constexpr auto name = "levenshtein";
 };
 
-template <typename LeftDataType, typename RightDataType>
-struct LevenshteinImpl {
-    using ResultDataType = DataTypeInt32;
-    using ResultPaddedPODArray = PaddedPODArray<Int32>;
+struct NameDamerauLevenshteinDistance {
+    static constexpr auto name = "damerau_levenshtein_distance";
+};
 
-    static Status vector_vector(const ColumnString::Chars& ldata,
-                                const ColumnString::Offsets& loffsets,
-                                const ColumnString::Chars& rdata,
-                                const ColumnString::Offsets& roffsets, ResultPaddedPODArray& res) {
-        DCHECK_EQ(loffsets.size(), roffsets.size());
+// 64MB limit for the distance matrix.
+static constexpr size_t MAX_DAMERAU_LEVENSHTEIN_MATRIX_CELLS = 16 * 1024 * 1024;
 
-        const size_t size = loffsets.size();
-        res.resize(size);
-        std::vector<size_t> left_offsets;
-        std::vector<size_t> right_offsets;
-        for (size_t i = 0; i < size; ++i) {
-            res[i] = levenshtein_distance(string_ref_at(ldata, loffsets, i),
-                                          string_ref_at(rdata, roffsets, i), left_offsets,
-                                          right_offsets);
-        }
-        return Status::OK();
+using Utf8Offsets = DorisVector<size_t>;
+
+static StringRef string_ref_at(const ColumnString::Chars& data,
+                               const ColumnString::Offsets& offsets, size_t i) {
+    DCHECK_LT(i, offsets.size());
+    const auto previous_offset = i == 0 ? 0 : offsets[i - 1];
+    return StringRef(data.data() + previous_offset, offsets[i] - previous_offset)
+            .trim_tail_padding_zero();
+}
+
+static void get_utf8_char_offsets(const StringRef& ref, Utf8Offsets& offsets) {
+    offsets.clear();
+    offsets.reserve(ref.size);
+    for (size_t i = 0, char_size = 0; i < ref.size; i += char_size) {
+        char_size = UTF8_BYTE_LENGTH[static_cast<unsigned char>(ref.data[i])];
+        offsets.push_back(i);
     }
+}
 
-    static Status vector_scalar(const ColumnString::Chars& ldata,
-                                const ColumnString::Offsets& loffsets, const StringRef& rdata,
-                                ResultPaddedPODArray& res) {
-        const size_t size = loffsets.size();
-        res.resize(size);
-        const auto right = rdata.trim_tail_padding_zero();
-        const bool right_ascii = simd::VStringFunctions::is_ascii(right);
-        std::vector<size_t> right_offsets;
-        simd::VStringFunctions::get_utf8_char_offsets(right, right_offsets);
-        std::vector<size_t> left_offsets;
-        for (size_t i = 0; i < size; ++i) {
-            res[i] = levenshtein_distance_with_right_offsets(string_ref_at(ldata, loffsets, i),
-                                                             left_offsets, right, right_offsets,
-                                                             right_ascii);
-        }
-        return Status::OK();
-    }
-
-    static Status scalar_vector(const StringRef& ldata, const ColumnString::Chars& rdata,
-                                const ColumnString::Offsets& roffsets, ResultPaddedPODArray& res) {
-        const size_t size = roffsets.size();
-        res.resize(size);
-        const auto left = ldata.trim_tail_padding_zero();
-        const bool left_ascii = simd::VStringFunctions::is_ascii(left);
-        std::vector<size_t> left_offsets;
-        simd::VStringFunctions::get_utf8_char_offsets(left, left_offsets);
-        std::vector<size_t> right_offsets;
-        for (size_t i = 0; i < size; ++i) {
-            res[i] = levenshtein_distance_with_left_offsets(left, left_offsets, left_ascii,
-                                                            string_ref_at(rdata, roffsets, i),
-                                                            right_offsets);
-        }
-        return Status::OK();
-    }
-
-private:
-    static StringRef string_ref_at(const ColumnString::Chars& data,
-                                   const ColumnString::Offsets& offsets, size_t i) {
-        DCHECK_LT(i, offsets.size());
-        const auto idx = static_cast<ssize_t>(i);
-        return StringRef(data.data() + offsets[idx - 1], offsets[idx] - offsets[idx - 1])
-                .trim_tail_padding_zero();
-    }
-
-    static Int32 levenshtein_distance_utf8(const StringRef& left,
-                                           const std::vector<size_t>& left_offsets,
-                                           const StringRef& right,
-                                           const std::vector<size_t>& right_offsets) {
+struct LevenshteinDistance {
+    static Status ascii(const StringRef& left, const StringRef& right, Int32& result) {
         const StringRef* left_ref = &left;
         const StringRef* right_ref = &right;
-        const std::vector<size_t>* left_offsets_ref = &left_offsets;
-        const std::vector<size_t>* right_offsets_ref = &right_offsets;
+        size_t m = left.size;
+        size_t n = right.size;
+
+        if (n > m) {
+            std::swap(left_ref, right_ref);
+            std::swap(m, n);
+        }
+
+        std::vector<Int32> prev(n + 1);
+        std::vector<Int32> curr(n + 1);
+        for (size_t j = 0; j <= n; ++j) {
+            prev[j] = static_cast<Int32>(j);
+        }
+
+        for (size_t i = 1; i <= m; ++i) {
+            curr[0] = static_cast<Int32>(i);
+            const char left_char = left_ref->data[i - 1];
+
+            for (size_t j = 1; j <= n; ++j) {
+                const Int32 cost = left_char == right_ref->data[j - 1] ? 0 : 1;
+                const Int32 insert_cost = curr[j - 1] + 1;
+                const Int32 delete_cost = prev[j] + 1;
+                const Int32 replace_cost = prev[j - 1] + cost;
+                curr[j] = std::min(std::min(insert_cost, delete_cost), replace_cost);
+            }
+            std::swap(prev, curr);
+        }
+
+        result = prev[n];
+        return Status::OK();
+    }
+
+    static Status utf8(const StringRef& left, const Utf8Offsets& left_offsets,
+                       const StringRef& right, const Utf8Offsets& right_offsets, Int32& result) {
+        const StringRef* left_ref = &left;
+        const StringRef* right_ref = &right;
+        const Utf8Offsets* left_offsets_ref = &left_offsets;
+        const Utf8Offsets* right_offsets_ref = &right_offsets;
         if (right_offsets_ref->size() > left_offsets_ref->size()) {
             std::swap(left_offsets_ref, right_offsets_ref);
             std::swap(left_ref, right_ref);
@@ -143,121 +142,333 @@ private:
             std::swap(prev, curr);
         }
 
-        return prev[n];
+        result = prev[n];
+        return Status::OK();
     }
+};
 
-    static Int32 levenshtein_distance_ascii(const StringRef& left, const StringRef& right) {
-        const StringRef* left_ref = &left;
-        const StringRef* right_ref = &right;
-        size_t m = left.size;
-        size_t n = right.size;
+struct DamerauLevenshteinDistance {
+    using SymbolId = Int32;
+    using SymbolVector = DorisVector<SymbolId>;
+    using SymbolMap =
+            std::unordered_map<std::string_view, SymbolId, std::hash<std::string_view>,
+                               std::equal_to<std::string_view>,
+                               CustomStdAllocator<std::pair<const std::string_view, SymbolId>>>;
 
-        if (n > m) {
-            std::swap(left_ref, right_ref);
-            std::swap(m, n);
+    // Do not use absl::strings_internal::CappedDamerauLevenshteinDistance here:
+    // 1. It is capped: distances greater than cutoff return cutoff + 1, not exact distance.
+    // 2. It returns uint8_t and is intended for short strings.
+    // 3. It implements the restricted/OSA variant, not full Damerau-Levenshtein.
+    // 4. It works on bytes, cannot handle UTF-8 characters.
+    static Status ascii(const StringRef& left, const StringRef& right, Int32& result) {
+        const size_t m = left.size;
+        const size_t n = right.size;
+        if (m == 0) {
+            result = static_cast<Int32>(n);
+            return Status::OK();
+        }
+        if (n == 0) {
+            result = static_cast<Int32>(m);
+            return Status::OK();
         }
 
-        std::vector<Int32> prev(n + 1);
-        std::vector<Int32> curr(n + 1);
+        size_t matrix_cells = 0;
+        RETURN_IF_ERROR(get_matrix_cell_count(m, n, matrix_cells));
+
+        const Int32 max_distance = static_cast<Int32>(m + n);
+        std::array<Int32, 256> last_row {};
+        PaddedPODArray<Int32> dist;
+        dist.resize_fill(matrix_cells, 0);
+        auto at = [&](size_t i, size_t j) -> Int32& { return dist[i * (n + 2) + j]; };
+
+        at(0, 0) = max_distance;
+        for (size_t i = 0; i <= m; ++i) {
+            at(i + 1, 0) = max_distance;
+            at(i + 1, 1) = static_cast<Int32>(i);
+        }
         for (size_t j = 0; j <= n; ++j) {
-            prev[j] = static_cast<Int32>(j);
+            at(0, j + 1) = max_distance;
+            at(1, j + 1) = static_cast<Int32>(j);
         }
 
         for (size_t i = 1; i <= m; ++i) {
-            curr[0] = static_cast<Int32>(i);
-            const char left_char = left_ref->data[i - 1];
-
+            Int32 last_match_col = 0;
+            const auto left_symbol = static_cast<unsigned char>(left.data[i - 1]);
             for (size_t j = 1; j <= n; ++j) {
-                const Int32 cost = left_char == right_ref->data[j - 1] ? 0 : 1;
-                const Int32 insert_cost = curr[j - 1] + 1;
-                const Int32 delete_cost = prev[j] + 1;
-                const Int32 replace_cost = prev[j - 1] + cost;
-                curr[j] = std::min(std::min(insert_cost, delete_cost), replace_cost);
+                const auto right_symbol = static_cast<unsigned char>(right.data[j - 1]);
+                const Int32 last_match_row = last_row[right_symbol];
+                const Int32 transposition_col = last_match_col;
+                Int32 cost = 1;
+                if (left_symbol == right_symbol) {
+                    cost = 0;
+                    last_match_col = static_cast<Int32>(j);
+                }
+
+                const Int32 replace_cost = at(i, j) + cost;
+                const Int32 insert_cost = at(i + 1, j) + 1;
+                const Int32 delete_cost = at(i, j + 1) + 1;
+                const Int32 transpose_cost = at(last_match_row, transposition_col) +
+                                             static_cast<Int32>(i - last_match_row - 1) + 1 +
+                                             static_cast<Int32>(j - transposition_col - 1);
+                at(i + 1, j + 1) = std::min(std::min(replace_cost, insert_cost),
+                                            std::min(delete_cost, transpose_cost));
             }
-            std::swap(prev, curr);
+            last_row[left_symbol] = static_cast<Int32>(i);
         }
 
-        return prev[n];
+        result = at(m + 1, n + 1);
+        return Status::OK();
     }
 
-    static Int32 levenshtein_distance(const StringRef& left, const StringRef& right,
-                                      std::vector<size_t>& left_offsets,
-                                      std::vector<size_t>& right_offsets) {
+    /*
+    * Keep the UTF-8 symbol mapping path. The benchmark below shows that mapping each UTF-8
+    * character to an integer symbol first is faster than comparing UTF-8 slices inside the
+    * dp loop, because the hot loop only needs integer equality and indexed last-row lookup.
+    *
+    * --------------------------------------------------------------------------
+    * Benchmark                                             Time             CPU
+    * --------------------------------------------------------------------------
+    * BM_DamerauLevenshtein_UTF8/SymbolMapped_16         1.85 us         1.82 us
+    * BM_DamerauLevenshtein_UTF8/DirectCompare_16        6.48 us         6.43 us
+    * BM_DamerauLevenshtein_UTF8/SymbolMapped_64         12.6 us         12.6 us
+    * BM_DamerauLevenshtein_UTF8/DirectCompare_64         119 us          118 us
+    * BM_DamerauLevenshtein_UTF8/SymbolMapped_128        43.4 us         43.1 us
+    * BM_DamerauLevenshtein_UTF8/DirectCompare_128        478 us          475 us
+    * BM_DamerauLevenshtein_UTF8/SymbolMapped_256         160 us          159 us
+    * BM_DamerauLevenshtein_UTF8/DirectCompare_256       1892 us         1882 us
+    */
+    static Status utf8(const StringRef& left, const Utf8Offsets& left_offsets,
+                       const StringRef& right, const Utf8Offsets& right_offsets, Int32& result) {
+        SymbolVector left_symbols;
+        SymbolVector right_symbols;
+        SymbolMap symbol_ids;
+        symbol_ids.reserve(left_offsets.size() + right_offsets.size());
+        SymbolId next_symbol = 0;
+        append_utf8_symbols(left, left_offsets, left_symbols, symbol_ids, next_symbol);
+        append_utf8_symbols(right, right_offsets, right_symbols, symbol_ids, next_symbol);
+        return by_symbols(left_symbols, right_symbols, next_symbol, result);
+    }
+
+private:
+    static Status by_symbols(const SymbolVector& left, const SymbolVector& right,
+                             size_t alphabet_size, Int32& result) {
+        const size_t m = left.size();
+        const size_t n = right.size();
+        if (m == 0) {
+            result = static_cast<Int32>(n);
+            return Status::OK();
+        }
+        if (n == 0) {
+            result = static_cast<Int32>(m);
+            return Status::OK();
+        }
+
+        size_t matrix_cells = 0;
+        RETURN_IF_ERROR(get_matrix_cell_count(m, n, matrix_cells));
+
+        const Int32 max_distance = static_cast<Int32>(m + n);
+        DorisVector<Int32> last_row(alphabet_size, 0);
+        PaddedPODArray<Int32> dist;
+        dist.resize_fill(matrix_cells, 0);
+        auto at = [&](size_t i, size_t j) -> Int32& { return dist[i * (n + 2) + j]; };
+
+        at(0, 0) = max_distance;
+        for (size_t i = 0; i <= m; ++i) {
+            at(i + 1, 0) = max_distance;
+            at(i + 1, 1) = static_cast<Int32>(i);
+        }
+        for (size_t j = 0; j <= n; ++j) {
+            at(0, j + 1) = max_distance;
+            at(1, j + 1) = static_cast<Int32>(j);
+        }
+
+        for (size_t i = 1; i <= m; ++i) {
+            Int32 last_match_col = 0;
+            for (size_t j = 1; j <= n; ++j) {
+                const Int32 last_match_row = last_row[right[j - 1]];
+                const Int32 transposition_col = last_match_col;
+                Int32 cost = 1;
+                if (left[i - 1] == right[j - 1]) {
+                    cost = 0;
+                    last_match_col = static_cast<Int32>(j);
+                }
+
+                const Int32 replace_cost = at(i, j) + cost;
+                const Int32 insert_cost = at(i + 1, j) + 1;
+                const Int32 delete_cost = at(i, j + 1) + 1;
+                const Int32 transpose_cost = at(last_match_row, transposition_col) +
+                                             static_cast<Int32>(i - last_match_row - 1) + 1 +
+                                             static_cast<Int32>(j - transposition_col - 1);
+                at(i + 1, j + 1) = std::min(std::min(replace_cost, insert_cost),
+                                            std::min(delete_cost, transpose_cost));
+            }
+            last_row[left[i - 1]] = static_cast<Int32>(i);
+        }
+
+        result = at(m + 1, n + 1);
+        return Status::OK();
+    }
+
+    static Status get_matrix_cell_count(size_t m, size_t n, size_t& matrix_cells) {
+        const auto max_size = std::numeric_limits<size_t>::max();
+        if (m > max_size - 2 || n > max_size - 2) {
+            return Status::InvalidArgument(
+                    "damerau_levenshtein_distance input is too large to allocate distance matrix");
+        }
+
+        const size_t rows = m + 2;
+        const size_t cols = n + 2;
+        if (rows > max_size / cols) {
+            return Status::InvalidArgument(
+                    "damerau_levenshtein_distance distance matrix size overflows");
+        }
+
+        matrix_cells = rows * cols;
+        if (matrix_cells > MAX_DAMERAU_LEVENSHTEIN_MATRIX_CELLS) {
+            return Status::InvalidArgument(
+                    "damerau_levenshtein_distance distance matrix is too large: {} cells exceeds "
+                    "limit {}",
+                    matrix_cells, MAX_DAMERAU_LEVENSHTEIN_MATRIX_CELLS);
+        }
+        return Status::OK();
+    }
+
+    static void append_utf8_symbols(const StringRef& str, const Utf8Offsets& offsets,
+                                    SymbolVector& symbols, SymbolMap& symbol_ids,
+                                    SymbolId& next_symbol) {
+        symbols.reserve(symbols.size() + offsets.size());
+        for (size_t i = 0; i < offsets.size(); ++i) {
+            const size_t offset = offsets[i];
+            const size_t next_offset = i + 1 < offsets.size() ? offsets[i + 1] : str.size;
+            const std::string_view symbol(str.data + offset, next_offset - offset);
+            auto [it, inserted] = symbol_ids.emplace(symbol, next_symbol);
+            if (inserted) {
+                ++next_symbol;
+            }
+            symbols.push_back(it->second);
+        }
+    }
+};
+
+template <typename Distance>
+struct StringDistanceImplBase {
+    using ResultDataType = DataTypeInt32;
+    using ResultPaddedPODArray = PaddedPODArray<Int32>;
+
+    static Status vector_vector(const ColumnString::Chars& ldata,
+                                const ColumnString::Offsets& loffsets,
+                                const ColumnString::Chars& rdata,
+                                const ColumnString::Offsets& roffsets, ResultPaddedPODArray& res) {
+        DCHECK_EQ(loffsets.size(), roffsets.size());
+
+        const size_t size = loffsets.size();
+        res.resize(size);
+        Utf8Offsets left_offsets;
+        Utf8Offsets right_offsets;
+        for (size_t i = 0; i < size; ++i) {
+            RETURN_IF_ERROR(distance(string_ref_at(ldata, loffsets, i),
+                                     string_ref_at(rdata, roffsets, i), left_offsets, right_offsets,
+                                     res[i]));
+        }
+        return Status::OK();
+    }
+
+    static Status vector_scalar(const ColumnString::Chars& data,
+                                const ColumnString::Offsets& offsets, const StringRef& constant,
+                                ResultPaddedPODArray& res) {
+        return vector_const(data, offsets, constant, res);
+    }
+
+    static Status scalar_vector(const StringRef& constant, const ColumnString::Chars& data,
+                                const ColumnString::Offsets& offsets, ResultPaddedPODArray& res) {
+        return vector_const(data, offsets, constant, res);
+    }
+
+private:
+    static Status vector_const(const ColumnString::Chars& data,
+                               const ColumnString::Offsets& offsets, const StringRef& constant,
+                               ResultPaddedPODArray& res) {
+        const size_t size = offsets.size();
+        res.resize(size);
+        const auto constant_ref = constant.trim_tail_padding_zero();
+        const bool constant_ascii = simd::VStringFunctions::is_ascii(constant_ref);
+        Utf8Offsets constant_offsets;
+        get_utf8_char_offsets(constant_ref, constant_offsets);
+        Utf8Offsets value_offsets;
+        for (size_t i = 0; i < size; ++i) {
+            RETURN_IF_ERROR(distance_with_const_offsets(string_ref_at(data, offsets, i),
+                                                        value_offsets, constant_ref,
+                                                        constant_offsets, constant_ascii, res[i]));
+        }
+        return Status::OK();
+    }
+
+    static Status distance(const StringRef& left, const StringRef& right, Utf8Offsets& left_offsets,
+                           Utf8Offsets& right_offsets, Int32& result) {
         const bool left_ascii = simd::VStringFunctions::is_ascii(left);
         const bool right_ascii = simd::VStringFunctions::is_ascii(right);
         if (left_ascii && right_ascii) {
-            return levenshtein_distance_ascii(left, right);
+            return Distance::ascii(left, right, result);
         }
 
         if (left.size == 0) {
-            return static_cast<Int32>(simd::VStringFunctions::get_char_len(right.data, right.size));
+            result = static_cast<Int32>(
+                    simd::VStringFunctions::get_char_len(right.data, right.size));
+            return Status::OK();
         }
         if (right.size == 0) {
-            return static_cast<Int32>(simd::VStringFunctions::get_char_len(left.data, left.size));
+            result = static_cast<Int32>(simd::VStringFunctions::get_char_len(left.data, left.size));
+            return Status::OK();
         }
 
-        simd::VStringFunctions::get_utf8_char_offsets(left, left_offsets);
-        simd::VStringFunctions::get_utf8_char_offsets(right, right_offsets);
-        return levenshtein_distance_utf8(left, left_offsets, right, right_offsets);
+        get_utf8_char_offsets(left, left_offsets);
+        get_utf8_char_offsets(right, right_offsets);
+        return Distance::utf8(left, left_offsets, right, right_offsets, result);
     }
 
-    static Int32 levenshtein_distance_with_right_offsets(const StringRef& left,
-                                                         std::vector<size_t>& left_offsets,
-                                                         const StringRef& right,
-                                                         const std::vector<size_t>& right_offsets,
-                                                         bool right_ascii) {
-        const bool left_ascii = simd::VStringFunctions::is_ascii(left);
-        if (left_ascii && right_ascii) {
-            return levenshtein_distance_ascii(left, right);
+    static Status distance_with_const_offsets(const StringRef& value, Utf8Offsets& value_offsets,
+                                              const StringRef& constant,
+                                              const Utf8Offsets& constant_offsets,
+                                              bool constant_ascii, Int32& result) {
+        const bool value_ascii = simd::VStringFunctions::is_ascii(value);
+        if (value_ascii && constant_ascii) {
+            return Distance::ascii(value, constant, result);
         }
 
-        if (left.size == 0) {
-            return static_cast<Int32>(right_offsets.size());
+        if (value.size == 0) {
+            result = static_cast<Int32>(constant_offsets.size());
+            return Status::OK();
         }
-        if (right.size == 0) {
-            return left_ascii ? static_cast<Int32>(left.size)
-                              : static_cast<Int32>(
-                                        simd::VStringFunctions::get_char_len(left.data, left.size));
+        if (constant.size == 0) {
+            result = value_ascii ? static_cast<Int32>(value.size)
+                                 : static_cast<Int32>(simd::VStringFunctions::get_char_len(
+                                           value.data, value.size));
+            return Status::OK();
         }
 
-        simd::VStringFunctions::get_utf8_char_offsets(left, left_offsets);
-        return levenshtein_distance_utf8(left, left_offsets, right, right_offsets);
+        get_utf8_char_offsets(value, value_offsets);
+        return Distance::utf8(value, value_offsets, constant, constant_offsets, result);
     }
+};
 
-    static Int32 levenshtein_distance_with_left_offsets(const StringRef& left,
-                                                        const std::vector<size_t>& left_offsets,
-                                                        bool left_ascii, const StringRef& right,
-                                                        std::vector<size_t>& right_offsets) {
-        const bool right_ascii = simd::VStringFunctions::is_ascii(right);
-        if (left_ascii && right_ascii) {
-            return levenshtein_distance_ascii(left, right);
-        }
+template <typename LeftDataType, typename RightDataType>
+struct LevenshteinImpl : public StringDistanceImplBase<LevenshteinDistance> {};
 
-        if (left.size == 0) {
-            return static_cast<Int32>(
-                    right_ascii ? right.size
-                                : simd::VStringFunctions::get_char_len(right.data, right.size));
-        }
-        if (right.size == 0) {
-            return static_cast<Int32>(left_offsets.size());
-        }
-
-        simd::VStringFunctions::get_utf8_char_offsets(right, right_offsets);
-        return levenshtein_distance_utf8(left, left_offsets, right, right_offsets);
-    }
-
-    static Int32 levenshtein_distance(const StringRef& left, const StringRef& right) {
-        std::vector<size_t> left_offsets;
-        std::vector<size_t> right_offsets;
-        return levenshtein_distance(left, right, left_offsets, right_offsets);
-    }
+template <typename LeftDataType, typename RightDataType>
+struct DamerauLevenshteinDistanceImpl : public StringDistanceImplBase<DamerauLevenshteinDistance> {
 };
 
 using FunctionLevenshtein =
         FunctionBinaryToType<DataTypeString, DataTypeString, LevenshteinImpl, NameLevenshtein>;
+using FunctionDamerauLevenshteinDistance =
+        FunctionBinaryToType<DataTypeString, DataTypeString, DamerauLevenshteinDistanceImpl,
+                             NameDamerauLevenshteinDistance>;
 
 void register_function_levenshtein(SimpleFunctionFactory& factory) {
     factory.register_function<FunctionLevenshtein>();
+    factory.register_alias(FunctionLevenshtein::name, "levenshtein_distance");
+    factory.register_alias(FunctionLevenshtein::name, "edit_distance");
+    factory.register_function<FunctionDamerauLevenshteinDistance>();
 }
 
 } // namespace doris

fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinScalarFunctions.java

@@ -163,6 +163,7 @@ import org.apache.doris.nereids.trees.expressions.functions.scalar.CurrentTime;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.CurrentUser;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.CutIpv6;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.CutToFirstSignificantSubdomain;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.DamerauLevenshteinDistance;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.Database;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.Date;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.DateDiff;
@@ -738,6 +739,7 @@ public class BuiltinScalarFunctions implements FunctionHelper {
             scalar(CurrentUser.class, "current_user"),
             scalar(CutIpv6.class, "cut_ipv6"),
             scalar(CutToFirstSignificantSubdomain.class, "cut_to_first_significant_subdomain"),
+            scalar(DamerauLevenshteinDistance.class, "damerau_levenshtein_distance"),
             scalar(Database.class, "database", "schema", "current_database"),
             scalar(Date.class, "date"),
             scalar(DateDiff.class, "datediff"),
@@ -897,9 +899,9 @@ public class BuiltinScalarFunctions implements FunctionHelper {
             scalar(LastQueryId.class, "last_query_id"),
             scalar(Lcm.class, "lcm"),
             scalar(Least.class, "least"),
-            scalar(Levenshtein.class, "levenshtein"),
             scalar(Left.class, "left", "strleft"),
             scalar(Length.class, "length", "octet_length"),
+            scalar(Levenshtein.class, "levenshtein", "levenshtein_distance", "edit_distance"),
             scalar(Crc32.class, "crc32"),
             scalar(Crc32Internal.class, "crc32_internal"),
             scalar(Like.class, "like"),

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/executable/StringArithmetic.java

@@ -20,6 +20,7 @@ package org.apache.doris.nereids.trees.expressions.functions.executable;
 import org.apache.doris.nereids.exceptions.AnalysisException;
 import org.apache.doris.nereids.exceptions.NotSupportedException;
 import org.apache.doris.nereids.trees.expressions.ExecFunction;
+import org.apache.doris.nereids.trees.expressions.ExecFunctionList;
 import org.apache.doris.nereids.trees.expressions.Expression;
 import org.apache.doris.nereids.trees.expressions.literal.ArrayLiteral;
 import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
@@ -52,8 +53,10 @@ import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.regex.Pattern;
 
 /**
@@ -61,6 +64,8 @@ import java.util.regex.Pattern;
  * concat
  */
 public class StringArithmetic {
+    private static final long MAX_DAMERAU_LEVENSHTEIN_MATRIX_CELLS = 16L * 1024L * 1024L;
+
     private static Literal castStringLikeLiteral(StringLikeLiteral first, String value) {
         if (first instanceof StringLiteral) {
             return new StringLiteral(value);
@@ -1217,7 +1222,11 @@ public class StringArithmetic {
     /**
      * Executable arithmetic functions levenshtein
      */
-    @ExecFunction(name = "levenshtein")
+    @ExecFunctionList({
+            @ExecFunction(name = "levenshtein"),
+            @ExecFunction(name = "levenshtein_distance"),
+            @ExecFunction(name = "edit_distance")
+    })
     public static Expression levenshtein(StringLikeLiteral first, StringLikeLiteral second) {
         int[] left = first.getValue().codePoints().toArray();
         int[] right = second.getValue().codePoints().toArray();
@@ -1261,6 +1270,73 @@ public class StringArithmetic {
         return new IntegerLiteral(prev[n]);
     }
 
+    /**
+     * Executable arithmetic functions damerau_levenshtein_distance
+     */
+    @ExecFunction(name = "damerau_levenshtein_distance")
+    public static Expression damerauLevenshteinDistance(StringLikeLiteral first,
+            StringLikeLiteral second) {
+        int[] left = first.getValue().codePoints().toArray();
+        int[] right = second.getValue().codePoints().toArray();
+        return new IntegerLiteral(damerauLevenshteinDistance(left, right));
+    }
+
+    private static int damerauLevenshteinDistance(int[] left, int[] right) {
+        int m = left.length;
+        int n = right.length;
+        if (m == 0) {
+            return n;
+        }
+        if (n == 0) {
+            return m;
+        }
+
+        long matrixCells = ((long) m + 2L) * ((long) n + 2L);
+        if (matrixCells > MAX_DAMERAU_LEVENSHTEIN_MATRIX_CELLS) {
+            throw new AnalysisException("damerau_levenshtein_distance distance matrix is too large: "
+                    + matrixCells + " cells exceeds limit " + MAX_DAMERAU_LEVENSHTEIN_MATRIX_CELLS);
+        }
+
+        int maxDistance = m + n;
+        int cols = n + 2;
+        int[] distance = new int[(int) matrixCells];
+        Map<Integer, Integer> lastRow = new HashMap<>();
+
+        distance[0] = maxDistance;
+        for (int i = 0; i <= m; i++) {
+            distance[(i + 1) * cols] = maxDistance;
+            distance[(i + 1) * cols + 1] = i;
+        }
+        for (int j = 0; j <= n; j++) {
+            distance[j + 1] = maxDistance;
+            distance[cols + j + 1] = j;
+        }
+
+        for (int i = 1; i <= m; i++) {
+            int lastMatchCol = 0;
+            for (int j = 1; j <= n; j++) {
+                int lastMatchRow = lastRow.getOrDefault(right[j - 1], 0);
+                int transpositionCol = lastMatchCol;
+                int cost = 1;
+                if (left[i - 1] == right[j - 1]) {
+                    cost = 0;
+                    lastMatchCol = j;
+                }
+
+                int replaceCost = distance[i * cols + j] + cost;
+                int insertCost = distance[(i + 1) * cols + j] + 1;
+                int deleteCost = distance[i * cols + j + 1] + 1;
+                int transposeCost = distance[lastMatchRow * cols + transpositionCol]
+                        + i - lastMatchRow - 1 + 1 + j - transpositionCol - 1;
+                distance[(i + 1) * cols + j + 1] = Math.min(Math.min(replaceCost, insertCost),
+                        Math.min(deleteCost, transposeCost));
+            }
+            lastRow.put(left[i - 1], i);
+        }
+
+        return distance[(m + 1) * cols + n + 1];
+    }
+
     /**
      * Executable arithmetic functions hamming_distance
      */

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/DamerauLevenshteinDistance.java

@@ -0,0 +1,76 @@
+// 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.doris.nereids.trees.expressions.functions.scalar;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.ExplicitlyCastableSignature;
+import org.apache.doris.nereids.trees.expressions.functions.PropagateNullable;
+import org.apache.doris.nereids.trees.expressions.shape.BinaryExpression;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.types.StringType;
+import org.apache.doris.nereids.types.VarcharType;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+/**
+ * ScalarFunction 'damerau_levenshtein_distance'.
+ */
+public class DamerauLevenshteinDistance extends ScalarFunction
+        implements BinaryExpression, ExplicitlyCastableSignature, PropagateNullable {
+
+    public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
+            FunctionSignature.ret(IntegerType.INSTANCE).args(VarcharType.SYSTEM_DEFAULT, VarcharType.SYSTEM_DEFAULT),
+            FunctionSignature.ret(IntegerType.INSTANCE).args(StringType.INSTANCE, StringType.INSTANCE)
+    );
+
+    /**
+     * constructor with 2 arguments.
+     */
+    public DamerauLevenshteinDistance(Expression arg0, Expression arg1) {
+        super("damerau_levenshtein_distance", arg0, arg1);
+    }
+
+    /** constructor for withChildren and reuse signature */
+    private DamerauLevenshteinDistance(ScalarFunctionParams functionParams) {
+        super(functionParams);
+    }
+
+    /**
+     * withChildren.
+     */
+    @Override
+    public DamerauLevenshteinDistance withChildren(List<Expression> children) {
+        Preconditions.checkArgument(children.size() == 2);
+        return new DamerauLevenshteinDistance(getFunctionParams(children));
+    }
+
+    @Override
+    public List<FunctionSignature> getSignatures() {
+        return SIGNATURES;
+    }
+
+    @Override
+    public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+        return visitor.visitDamerauLevenshteinDistance(this, context);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/ScalarFunctionVisitor.java

@@ -173,6 +173,7 @@ import org.apache.doris.nereids.trees.expressions.functions.scalar.CurrentTime;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.CurrentUser;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.CutIpv6;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.CutToFirstSignificantSubdomain;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.DamerauLevenshteinDistance;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.Database;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.Date;
 import org.apache.doris.nereids.trees.expressions.functions.scalar.DateDiff;
@@ -1269,6 +1270,11 @@ public interface ScalarFunctionVisitor<R, C> {
         return visitScalarFunction(daysSub, context);
     }
 
+    default R visitDamerauLevenshteinDistance(DamerauLevenshteinDistance damerauLevenshteinDistance,
+            C context) {
+        return visitScalarFunction(damerauLevenshteinDistance, context);
+    }
+
     default R visitDictGet(DictGet dictGet, C context) {
         return visitScalarFunction(dictGet, context);
     }

regression-test/data/query_p0/sql_functions/string_functions/test_string_all.out

@@ -977,6 +977,15 @@ R163	R163
 -- !levenshtein_334 --
 \N	2	1
 
+-- !levenshtein_alias_335 --
+2	3
+
+-- !damerau_levenshtein_336 --
+0	3	2	1	1	1	1	2
+
+-- !damerau_levenshtein_337 --
+\N	\N	3
+
 -- !levenshtein_tbl --
 1	3
 2	0
@@ -987,6 +996,19 @@ R163	R163
 7	3
 8	2
 9	1
+10	3
+
+-- !damerau_levenshtein_tbl --
+1	3
+2	0
+3	1
+4	\N
+5	1
+6	1
+7	3
+8	2
+9	1
+10	2
 
 -- !levenshtein_nn_vector_vector --
 1	1
@@ -1006,6 +1028,26 @@ R163	R163
 -- !levenshtein_nn_scalar_vector_utf8 --
 3	1
 
+-- !damerau_levenshtein_nn_vector_vector --
+1	1
+2	4
+3	1
+4	1
+
+-- !damerau_levenshtein_nn_vector_scalar_ascii --
+1	1
+2	1
+
+-- !damerau_levenshtein_nn_scalar_vector_ascii --
+1	1
+2	1
+
+-- !damerau_levenshtein_nn_vector_scalar_utf8 --
+4	1
+
+-- !damerau_levenshtein_nn_scalar_vector_utf8 --
+4	1
+
 -- !levenshtein_vector_scalar_nullable --
 2	0
 4	\N

regression-test/suites/query_p0/sql_functions/string_functions/test_string_all.groovy

@@ -774,6 +774,16 @@ suite("string_functions_all") {
     testFoldConst("SELECT levenshtein('abcd', 'abdc'), levenshtein('你好呀', '你好'), levenshtein('a你b', 'a们b');")
     qt_levenshtein_334 "SELECT levenshtein(NULL, NULL), levenshtein('', '你好'), levenshtein('你好世界', '你好世间');"
     testFoldConst("SELECT levenshtein(NULL, NULL), levenshtein('', '你好'), levenshtein('你好世界', '你好世间');")
+    qt_levenshtein_alias_335 "SELECT levenshtein_distance('abcd', 'abdc'), edit_distance('kitten', 'sitting');"
+    testFoldConst("SELECT levenshtein_distance('abcd', 'abdc'), edit_distance('kitten', 'sitting');")
+    qt_damerau_levenshtein_336 "SELECT damerau_levenshtein_distance('', ''), damerau_levenshtein_distance('kitten', 'sitting'), damerau_levenshtein_distance('flaw', 'lawn'), damerau_levenshtein_distance('abcd', 'abdc'), damerau_levenshtein_distance('ca', 'ac'), damerau_levenshtein_distance('你好', '好你'), damerau_levenshtein_distance('数据库', '数据'), damerau_levenshtein_distance('CA', 'ABC');"
+    testFoldConst("SELECT damerau_levenshtein_distance('', ''), damerau_levenshtein_distance('kitten', 'sitting'), damerau_levenshtein_distance('flaw', 'lawn'), damerau_levenshtein_distance('abcd', 'abdc'), damerau_levenshtein_distance('ca', 'ac'), damerau_levenshtein_distance('你好', '好你'), damerau_levenshtein_distance('数据库', '数据'), damerau_levenshtein_distance('CA', 'ABC');")
+    qt_damerau_levenshtein_337 "SELECT damerau_levenshtein_distance(NULL, 'abc'), damerau_levenshtein_distance('abc', NULL), damerau_levenshtein_distance('', 'abc');"
+    testFoldConst("SELECT damerau_levenshtein_distance(NULL, 'abc'), damerau_levenshtein_distance('abc', NULL), damerau_levenshtein_distance('', 'abc');")
+    test {
+        sql "SELECT damerau_levenshtein_distance(repeat('a', 4096), repeat('b', 4096));"
+        exception "damerau_levenshtein_distance distance matrix is too large"
+    }
     sql """DROP TABLE IF EXISTS string_distance_lv_test"""
     sql """
         CREATE TABLE IF NOT EXISTS string_distance_lv_test (
@@ -794,9 +804,26 @@ suite("string_functions_all") {
         (6, 'abcd', 'abdc'),
         (7, '', '数据库'),
         (8, '你好', ''),
-        (9, '数据', '数据库')
+        (9, '数据', '数据库'),
+        (10, 'CA', 'ABC')
     """
     qt_levenshtein_tbl "SELECT id, levenshtein(s1, s2) FROM string_distance_lv_test ORDER BY id"
+    qt_damerau_levenshtein_tbl "SELECT id, damerau_levenshtein_distance(s1, s2) FROM string_distance_lv_test ORDER BY id"
+    sql """DROP TABLE IF EXISTS string_distance_big_damerau_test"""
+    sql """
+        CREATE TABLE IF NOT EXISTS string_distance_big_damerau_test (
+            id int,
+            s1 VARCHAR(5000),
+            s2 VARCHAR(5000)
+        )
+        DISTRIBUTED BY HASH(id) BUCKETS 1
+        PROPERTIES ("replication_num"="1")
+    """
+    sql """INSERT INTO string_distance_big_damerau_test VALUES (1, repeat('a', 4096), repeat('b', 4096))"""
+    test {
+        sql "SELECT damerau_levenshtein_distance(s1, s2) FROM string_distance_big_damerau_test;"
+        exception "damerau_levenshtein_distance distance matrix is too large"
+    }
 
     sql """DROP TABLE IF EXISTS string_distance_nn_test"""
     sql """
@@ -820,6 +847,11 @@ suite("string_functions_all") {
     qt_levenshtein_nn_scalar_vector_ascii "SELECT id, levenshtein('abc', s1) FROM string_distance_nn_test WHERE id = 1 ORDER BY id"
     qt_levenshtein_nn_vector_scalar_utf8 "SELECT id, levenshtein(s1, '你们') FROM string_distance_nn_test WHERE id = 3 ORDER BY id"
     qt_levenshtein_nn_scalar_vector_utf8 "SELECT id, levenshtein('你们', s1) FROM string_distance_nn_test WHERE id = 3 ORDER BY id"
+    qt_damerau_levenshtein_nn_vector_vector "SELECT id, damerau_levenshtein_distance(s1, s2) FROM string_distance_nn_test ORDER BY id"
+    qt_damerau_levenshtein_nn_vector_scalar_ascii "SELECT id, damerau_levenshtein_distance(s1, 'abdc') FROM string_distance_nn_test WHERE id IN (1, 2) ORDER BY id"
+    qt_damerau_levenshtein_nn_scalar_vector_ascii "SELECT id, damerau_levenshtein_distance('abdc', s1) FROM string_distance_nn_test WHERE id IN (1, 2) ORDER BY id"
+    qt_damerau_levenshtein_nn_vector_scalar_utf8 "SELECT id, damerau_levenshtein_distance(s1, '数库据') FROM string_distance_nn_test WHERE id = 4 ORDER BY id"
+    qt_damerau_levenshtein_nn_scalar_vector_utf8 "SELECT id, damerau_levenshtein_distance('数库据', s1) FROM string_distance_nn_test WHERE id = 4 ORDER BY id"
     qt_levenshtein_vector_scalar_nullable "SELECT id, levenshtein(s1, 'abc') FROM string_distance_lv_test WHERE id IN (2, 4) ORDER BY id"
     qt_levenshtein_scalar_vector_nullable "SELECT id, levenshtein('abc', s1) FROM string_distance_lv_test WHERE id IN (2, 4) ORDER BY id"
     qt_levenshtein_vector_scalar_utf8 "SELECT id, levenshtein(s1, '你好') FROM string_distance_lv_test WHERE id IN (5, 7) ORDER BY id"