feat(iceberg): Support Iceberg hash (#14025)

Summary:
The iceberg hash use mumur3 hash, which aligns with https://github.com/aappleby/smhasher/blob/master/src/MurmurHash3.cpp, firstly, process every 4 bytes as a chunk, then process remaining bytes by XOR, sparksql also uses this hash algorithm but is different with processing remaining bytes, which combine the remaining bytes. Extract the common function hashInt64 to functions/lib.

This class will be used for iceberg bucket transform and bucket function.
The iceberg mumur3 hash should be strictly with java implementation, then write by iceberg could read with iceberg Java, and the function call can also get the correct result.
The iceberg utility lib `velox_functions_iceberg_hash` will be linked by iceberg connector write to do partition transform. #13874

Pull Request resolved: #14025

Reviewed By: pedroerp

Differential Revision: D79732785

Pulled By: kgpai

fbshipit-source-id: 6122b94673f015dca5c8484722926709a30fe65e

Author: jinchengchenghh

CMakeLists.txt

@@ -131,6 +131,7 @@ option(VELOX_ENABLE_HIVE_CONNECTOR "Build Hive connector." ON)
 option(VELOX_ENABLE_TPCH_CONNECTOR "Build TPC-H connector." ON)
 option(VELOX_ENABLE_PRESTO_FUNCTIONS "Build Presto SQL functions." ON)
 option(VELOX_ENABLE_SPARK_FUNCTIONS "Build Spark SQL functions." ON)
+option(VELOX_ENABLE_ICEBERG_FUNCTIONS "Build Iceberg functions." ON)
 option(VELOX_ENABLE_EXPRESSION "Build expression." ON)
 option(
   VELOX_ENABLE_EXAMPLES

velox/functions/CMakeLists.txt

@@ -37,3 +37,7 @@ endif()
 if(${VELOX_BUILD_TESTING})
   add_subdirectory(tests)
 endif()
+
+if(${VELOX_ENABLE_ICEBERG_FUNCTIONS})
+  add_subdirectory(iceberg)
+endif()

velox/functions/iceberg/CMakeLists.txt

@@ -0,0 +1,22 @@
+# Copyright (c) Facebook, Inc. and its affiliates.
+#
+# Licensed 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.
+
+velox_add_library(velox_functions_iceberg_hash Murmur3Hash32.cpp)
+
+velox_link_libraries(velox_functions_iceberg_hash velox_common_base
+                     velox_functions_util)
+
+if(${VELOX_BUILD_TESTING})
+  add_subdirectory(tests)
+endif()

velox/functions/iceberg/Murmur3Hash32.cpp

@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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 "velox/functions/iceberg/Murmur3Hash32.h"
+#include "velox/common/base/BitUtil.h"
+
+namespace facebook::velox::functions::iceberg {
+
+namespace {
+
+constexpr int32_t kSeed = 0;
+} // namespace
+
+int32_t Murmur3Hash32::hashInt64(uint64_t input) {
+  return Murmur3Hash32Base::hashInt64(input, kSeed);
+}
+
+int32_t Murmur3Hash32::hashBytes(const char* input, uint32_t len) {
+  const uint8_t* data = reinterpret_cast<const uint8_t*>(input);
+  const int32_t nblocks = len / 4;
+
+  uint32_t h1 = kSeed;
+
+  // Process 4-byte chunks.
+  for (auto i = 0; i < nblocks; i++) {
+    uint32_t k1 = *reinterpret_cast<const uint32_t*>(data + i * 4);
+    k1 = mixK1(k1);
+    h1 = mixH1(h1, k1);
+  }
+
+  // Process remaining bytes.
+  const uint8_t* tail = data + nblocks * 4;
+  uint32_t k1 = 0;
+  switch (len & 3) {
+    case 3:
+      k1 ^= static_cast<uint32_t>(tail[2]) << 16;
+      [[fallthrough]];
+    case 2:
+      k1 ^= static_cast<uint32_t>(tail[1]) << 8;
+      [[fallthrough]];
+    case 1:
+      k1 ^= static_cast<uint32_t>(tail[0]);
+      h1 ^= mixK1(k1);
+  }
+
+  return fmix(h1, len);
+}
+
+} // namespace facebook::velox::functions::iceberg

velox/functions/iceberg/Murmur3Hash32.h

@@ -0,0 +1,42 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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.
+ */
+#pragma once
+
+#include <cstdint>
+#include "velox/functions/lib/Murmur3Hash32Base.h"
+
+namespace facebook::velox::functions::iceberg {
+
+/// This implementation should align with Iceberg bucket transform hash
+/// algorithm which uses Guava hash
+/// https://github.com/google/guava/blob/master/guava/src/com/google/common/hash/Murmur3_32HashFunction.java,
+/// Bucket partition transforms use a 32-bit hash of the source value. The
+/// 32-bit hash implementation is the 32-bit Murmur3 hash, x86 variant, seeded
+/// with 0. If not same with Iceberg java version, the partition will be
+/// different and can not read with Iceberg java.
+class Murmur3Hash32 : Murmur3Hash32Base {
+ public:
+  /// Value of type INTEGER and BIGINT is treated as unsigned type.
+  /// For the schema evolution, promote int to int64, treat int32 as uint64.
+  static int32_t hashInt64(uint64_t input);
+
+  /// Hash the bytes every 4 bytes, XOR on remaining bytes. Processing for the
+  /// remaining bytes is different with Spark murmur3 which combine with the
+  /// remaining bytes.
+  static int32_t hashBytes(const char* input, uint32_t len);
+};
+
+} // namespace facebook::velox::functions::iceberg

velox/functions/iceberg/tests/CMakeLists.txt

@@ -0,0 +1,26 @@
+# Copyright (c) Facebook, Inc. and its affiliates.
+#
+# Licensed 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.
+add_executable(velox_functions_iceberg_test Murmur3Hash32Test.cpp)
+
+add_test(
+  NAME velox_functions_iceberg_test
+  COMMAND velox_functions_iceberg_test
+  WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
+
+target_link_libraries(
+  velox_functions_iceberg_test
+  velox_functions_iceberg_hash
+  GTest::gtest
+  GTest::gtest_main
+  GTest::gmock)

velox/functions/iceberg/tests/Murmur3Hash32Test.cpp

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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 "velox/functions/iceberg/Murmur3Hash32.h"
+
+#include <gtest/gtest.h>
+
+namespace facebook::velox::functions::iceberg {
+namespace {
+
+TEST(Murmur3Hash32Test, bigint) {
+  Murmur3Hash32 func;
+  EXPECT_EQ(func.hashInt64(10), -289985220);
+  EXPECT_EQ(func.hashInt64(0), 1669671676);
+  EXPECT_EQ(func.hashInt64(-5), 1222806974);
+}
+
+TEST(Murmur3Hash32Test, string) {
+  Murmur3Hash32 func;
+
+  const auto hash = [&](std::string input) {
+    return func.hashBytes(input.c_str(), input.size());
+  };
+
+  EXPECT_EQ(hash("abcdefg"), -2009294074);
+  EXPECT_EQ(hash("abc"), -1277324294);
+  EXPECT_EQ(hash("abcd"), 1139631978);
+  EXPECT_EQ(hash("abcde"), -392455434);
+  EXPECT_EQ(hash("测试"), -25843656);
+  EXPECT_EQ(hash("测试raul试测"), -912788207);
+  EXPECT_EQ(hash(""), 0);
+  EXPECT_EQ(hash("Товары"), 1817480714);
+  EXPECT_EQ(hash("😀"), -1095487750);
+}
+} // namespace
+} // namespace facebook::velox::functions::iceberg

velox/functions/lib/CMakeLists.txt

@@ -16,8 +16,8 @@ velox_add_library(velox_is_null_functions IsNull.cpp)
 
 velox_link_libraries(velox_is_null_functions velox_expression)
 
-velox_add_library(velox_functions_util LambdaFunctionUtil.cpp
-                  RowsTranslationUtil.cpp)
+velox_add_library(velox_functions_util Murmur3Hash32Base.cpp
+                  LambdaFunctionUtil.cpp RowsTranslationUtil.cpp)
 
 velox_link_libraries(velox_functions_util velox_vector velox_common_base)
 

velox/functions/lib/Murmur3Hash32Base.cpp

@@ -0,0 +1,58 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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 "velox/functions/lib/Murmur3Hash32Base.h"
+#include "velox/common/base/BitUtil.h"
+
+namespace facebook::velox::functions {
+
+uint32_t Murmur3Hash32Base::hashInt64(uint64_t input, uint32_t seed) {
+  uint32_t low = input;
+  uint32_t high = input >> 32;
+
+  uint32_t k1 = mixK1(low);
+  uint32_t h1 = mixH1(seed, k1);
+
+  k1 = mixK1(high);
+  h1 = mixH1(h1, k1);
+
+  return fmix(h1, 8);
+}
+
+uint32_t Murmur3Hash32Base::mixK1(uint32_t k1) {
+  k1 *= 0xcc9e2d51;
+  k1 = bits::rotateLeft(k1, 15);
+  k1 *= 0x1b873593;
+  return k1;
+}
+
+uint32_t Murmur3Hash32Base::mixH1(uint32_t h1, uint32_t k1) {
+  h1 ^= k1;
+  h1 = bits::rotateLeft(h1, 13);
+  h1 = h1 * 5 + 0xe6546b64;
+  return h1;
+}
+
+uint32_t Murmur3Hash32Base::fmix(uint32_t h1, uint32_t length) {
+  h1 ^= length;
+  h1 ^= h1 >> 16;
+  h1 *= 0x85ebca6b;
+  h1 ^= h1 >> 13;
+  h1 *= 0xc2b2ae35;
+  h1 ^= h1 >> 16;
+  return h1;
+}
+} // namespace facebook::velox::functions

velox/functions/lib/Murmur3Hash32Base.h

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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.
+ */
+#pragma once
+
+#include <cstdint>
+
+namespace facebook::velox::functions {
+/// Murmur3 aligns with Austin Appleby
+/// https://github.com/aappleby/smhasher/blob/master/src/MurmurHash3.cpp
+///
+/// Signed integer types have been remapped to unsigned types (as in the
+/// original) to avoid undefined signed integer overflow and sign extension.
+class Murmur3Hash32Base {
+ protected:
+  /// Hash the lower int, then combine with higher int, is a fast path of
+  /// hashBytes.
+  static uint32_t hashInt64(uint64_t input, uint32_t seed);
+
+  static uint32_t mixK1(uint32_t k1);
+
+  static uint32_t mixH1(uint32_t h1, uint32_t k1);
+
+  // Finalization mix - force all bits of a hash block to avalanche.
+  static uint32_t fmix(uint32_t h1, uint32_t length);
+};
+} // namespace facebook::velox::functions