Merge branch 'main' of github.com:facebookincubator/velox into joinFuzzer

Author: xin-zhang2

.github/workflows/scheduled.yml

@@ -318,7 +318,7 @@ jobs:
         uses: actions/upload-artifact@v4
         with:
           name: cache_fuzzer
-          path: velox/_build/debug/velox/exec/tests/velox_cache_fuzzer_test
+          path: velox/_build/debug/velox/exec/fuzzer/velox_cache_fuzzer
           retention-days: "${{ env.RETENTION }}"
 
       - name: Upload table evolution fuzzer
@@ -751,6 +751,8 @@ jobs:
     container: ghcr.io/facebookincubator/velox-dev:centos9
     needs: compile
     timeout-minutes: 120
+    # Temporarily disable on PRs till flakiness is fixed #12167
+    if: ${{ github.event_name != 'pull_request' }}
     steps:
 
       - name: Download cache fuzzer
@@ -760,24 +762,24 @@ jobs:
 
       - name: Run Cache Fuzzer
         run: |
-          mkdir -p /tmp/cache_fuzzer_test/logs/
-          chmod -R 777 /tmp/cache_fuzzer_test
-          chmod +x velox_cache_fuzzer_test
-          ./velox_cache_fuzzer_test \
+          mkdir -p /tmp/cache_fuzzer/logs/
+          chmod -R 777 /tmp/cache_fuzzer
+          chmod +x velox_cache_fuzzer
+          ./velox_cache_fuzzer \
                 --seed ${RANDOM} \
                 --duration_sec $DURATION \
                 --minloglevel=0 \
                 --stderrthreshold=2 \
-                --log_dir=/tmp/cache_fuzzer_test/logs \
+                --log_dir=/tmp/cache_fuzzer/logs \
             && echo -e "\n\Cache fuzzer run finished successfully."
 
       - name: Archive Cache production artifacts
         if: ${{ !cancelled() }}
         uses: actions/upload-artifact@v4
         with:
-            name: cache-fuzzer-test-logs
+            name: cache-fuzzer-logs
             path: |
-              /tmp/cache_fuzzer_test
+              /tmp/cache_fuzzer
 
   table-evolution-fuzzer-run:
     name: Table Evolution Fuzzer

CONTRIBUTING.md

@@ -287,7 +287,7 @@ with a benchmark.
    line arguments.
 
    ```
-# Test the new function in isolation. Use --only flag to restrict the set of functions
+   # Test the new function in isolation. Use --only flag to restrict the set of functions
    # and run for 60 seconds or longer.
    velox_expression_fuzzer_test --only <my-new-function-name> --duration_sec 60 --logtostderr=1 --enable_variadic_signatures --velox_fuzzer_enable_complex_types --velox_fuzzer_enable_decimal_type --lazy_vector_generation_ratio 0.2 --velox_fuzzer_enable_column_reuse --velox_fuzzer_enable_expression_reuse
 

velox/common/base/CMakeLists.txt

@@ -56,8 +56,7 @@ velox_link_libraries(
   velox_process
   glog::glog
   Folly::folly
-  fmt::fmt
-  gflags::gflags)
+  fmt::fmt)
 
 velox_add_library(velox_status Status.cpp)
 velox_link_libraries(

velox/common/caching/CMakeLists.txt

@@ -32,7 +32,6 @@ velox_link_libraries(
          velox_time
          Folly::folly
          fmt::fmt
-         gflags::gflags
   PRIVATE velox_time)
 
 if(${VELOX_BUILD_TESTING})

velox/common/caching/SsdFile.cpp

@@ -22,6 +22,7 @@
 #include "velox/common/base/SuccinctPrinter.h"
 #include "velox/common/caching/FileIds.h"
 #include "velox/common/caching/SsdCache.h"
+#include "velox/common/config/GlobalConfig.h"
 #include "velox/common/memory/Memory.h"
 #include "velox/common/process/TraceContext.h"
 
@@ -34,9 +35,6 @@
 #include <sys/types.h>
 #include <numeric>
 
-DEFINE_bool(ssd_odirect, true, "Use O_DIRECT for SSD cache IO");
-DEFINE_bool(ssd_verify_write, false, "Read back data after writing to SSD");
-
 namespace facebook::velox::cache {
 
 namespace {
@@ -118,7 +116,7 @@ SsdFile::SsdFile(const Config& config)
   process::TraceContext trace("SsdFile::SsdFile");
   filesystems::FileOptions fileOptions;
   fileOptions.shouldThrowOnFileAlreadyExists = false;
-  fileOptions.bufferIo = !FLAGS_ssd_odirect;
+  fileOptions.bufferIo = !config::globalConfig().useSsdODirect;
   writeFile_ = fs_->openFileForWrite(fileName_, fileOptions);
   readFile_ = fs_->openFileForRead(fileName_, fileOptions);
 
@@ -417,7 +415,7 @@ void SsdFile::write(std::vector<CachePin>& pins) {
           checksum = checksumEntry(*entry);
         }
         entries_[std::move(key)] = SsdRun(offset, size, checksum);
-        if (FLAGS_ssd_verify_write) {
+        if (config::globalConfig().verifySsdWrite) {
           verifyWrite(*entry, SsdRun(offset, size, checksum));
         }
         offset += size;

velox/common/caching/SsdFile.h

@@ -25,9 +25,6 @@
 #include "velox/common/file/FileInputStream.h"
 #include "velox/common/file/FileSystems.h"
 
-DECLARE_bool(ssd_odirect);
-DECLARE_bool(ssd_verify_write);
-
 namespace facebook::velox::cache {
 
 namespace test {

velox/common/caching/tests/AsyncDataCacheTest.cpp

@@ -27,6 +27,7 @@
 #include "velox/common/testutil/ScopedTestTime.h"
 #include "velox/common/testutil/TestValue.h"
 #include "velox/exec/tests/utils/TempDirectoryPath.h"
+#include "velox/flag_definitions/flags.h"
 
 #include <folly/executors/IOThreadPoolExecutor.h>
 #include <folly/executors/QueuedImmediateExecutor.h>
@@ -136,7 +137,8 @@ class AsyncDataCacheTest : public ::testing::TestWithParam<TestParam> {
     std::unique_ptr<SsdCache> ssdCache;
     if (ssdBytes > 0) {
       // tmpfs does not support O_DIRECT, so turn this off for testing.
-      FLAGS_ssd_odirect = false;
+      FLAGS_velox_ssd_odirect = false;
+      translateFlagsToGlobalConfig();
       // Make a new tempDirectory only if one is not already set. The
       // second creation of cache must find the checkpoint of the
       // previous one.
@@ -846,7 +848,8 @@ TEST_P(AsyncDataCacheTest, DISABLED_ssd) {
 
   // Read back all writes. This increases the chance of writes falling behind
   // new entry creation.
-  FLAGS_ssd_verify_write = true;
+  FLAGS_velox_ssd_verify_write = true;
+  translateFlagsToGlobalConfig();
 
   // We read kSsdBytes worth of data on 16 threads. The same data will be hit by
   // all threads. The expectation is that most of the data ends up on SSD. All
@@ -861,7 +864,8 @@ TEST_P(AsyncDataCacheTest, DISABLED_ssd) {
   ASSERT_LE(kRamBytes, ssdStats.bytesWritten);
 
   // We allow writes to proceed faster.
-  FLAGS_ssd_verify_write = false;
+  FLAGS_velox_ssd_verify_write = false;
+  translateFlagsToGlobalConfig();
   // We read the data back. The verify hook checks correct values. Error every
   // 13 batch loads.
   runThreads(16, [&](int32_t /*i*/) { loadLoop(0, kSsdBytes, 13); });

velox/common/caching/tests/CMakeLists.txt

@@ -39,6 +39,7 @@ target_link_libraries(
     velox_file_test_utils
     velox_memory
     velox_temp_path
+    velox_flag_definitions
     Folly::folly
     glog::glog
     GTest::gtest

velox/common/caching/tests/SsdFileTest.cpp

@@ -17,6 +17,7 @@
 #include "velox/common/base/tests/GTestUtils.h"
 #include "velox/common/caching/FileIds.h"
 #include "velox/common/caching/tests/CacheTestUtil.h"
+#include "velox/common/config/GlobalConfig.h"
 #include "velox/common/file/FileSystems.h"
 #include "velox/common/file/tests/FaultyFileSystem.h"
 #include "velox/common/memory/Memory.h"
@@ -73,7 +74,7 @@ class SsdFileTest : public testing::Test {
       bool disableFileCow = false,
       bool enableFaultInjection = false) {
     // tmpfs does not support O_DIRECT, so turn this off for testing.
-    FLAGS_ssd_odirect = false;
+    config::globalConfig().useSsdODirect = false;
     cache_ = AsyncDataCache::create(memory::memoryManager()->allocator());
     cacheHelper_ =
         std::make_unique<test::AsyncDataCacheTestHelper>(cache_.get());
@@ -327,7 +328,7 @@ TEST_F(SsdFileTest, writeAndRead) {
   constexpr int64_t kSsdSize = 16 * SsdFile::kRegionSize;
   std::vector<TestEntry> allEntries;
   initializeCache(kSsdSize);
-  FLAGS_ssd_verify_write = true;
+  config::globalConfig().verifySsdWrite = true;
   for (auto startOffset = 0; startOffset <= kSsdSize - SsdFile::kRegionSize;
        startOffset += SsdFile::kRegionSize) {
     auto pins =
@@ -402,7 +403,7 @@ TEST_F(SsdFileTest, checkpoint) {
   constexpr int64_t kSsdSize = 16 * SsdFile::kRegionSize;
   const uint64_t checkpointIntervalBytes = 5 * SsdFile::kRegionSize;
   const auto fileNameAlt = StringIdLease(fileIds(), "fileInStorageAlt");
-  FLAGS_ssd_verify_write = true;
+  config::globalConfig().verifySsdWrite = true;
   initializeCache(kSsdSize, checkpointIntervalBytes);
 
   std::vector<TestEntry> allEntries;
@@ -498,7 +499,7 @@ TEST_F(SsdFileTest, checkpoint) {
 TEST_F(SsdFileTest, fileCorruption) {
   constexpr int64_t kSsdSize = 16 * SsdFile::kRegionSize;
   const uint64_t checkpointIntervalBytes = 5 * SsdFile::kRegionSize;
-  FLAGS_ssd_verify_write = true;
+  config::globalConfig().verifySsdWrite = true;
 
   const auto populateCache = [&](std::vector<TestEntry>& entries) {
     entries.clear();
@@ -570,7 +571,7 @@ TEST_F(SsdFileTest, fileCorruption) {
 TEST_F(SsdFileTest, recoverFromCheckpointWithChecksum) {
   constexpr int64_t kSsdSize = 4 * SsdFile::kRegionSize;
   const uint64_t checkpointIntervalBytes = 3 * SsdFile::kRegionSize;
-  FLAGS_ssd_verify_write = true;
+  config::globalConfig().verifySsdWrite = true;
 
   // Test if cache data can be recovered with different settings.
   struct {

velox/common/compression/Compression.cpp

@@ -25,36 +25,36 @@ std::unique_ptr<folly::compression::Codec> compressionKindToCodec(
     CompressionKind kind) {
   switch (static_cast<int32_t>(kind)) {
     case CompressionKind_NONE:
-      return getCodec(folly::io::CodecType::NO_COMPRESSION);
+      return getCodec(folly::compression::CodecType::NO_COMPRESSION);
     case CompressionKind_ZLIB:
-      return getCodec(folly::io::CodecType::ZLIB);
+      return getCodec(folly::compression::CodecType::ZLIB);
     case CompressionKind_SNAPPY:
-      return getCodec(folly::io::CodecType::SNAPPY);
+      return getCodec(folly::compression::CodecType::SNAPPY);
     case CompressionKind_ZSTD:
-      return getCodec(folly::io::CodecType::ZSTD);
+      return getCodec(folly::compression::CodecType::ZSTD);
     case CompressionKind_LZ4:
-      return getCodec(folly::io::CodecType::LZ4);
+      return getCodec(folly::compression::CodecType::LZ4);
     case CompressionKind_GZIP:
-      return getCodec(folly::io::CodecType::GZIP);
+      return getCodec(folly::compression::CodecType::GZIP);
     default:
       VELOX_UNSUPPORTED(
           "Not support {} in folly", compressionKindToString(kind));
   }
 }
 
-CompressionKind codecTypeToCompressionKind(folly::io::CodecType type) {
+CompressionKind codecTypeToCompressionKind(folly::compression::CodecType type) {
   switch (type) {
-    case folly::io::CodecType::NO_COMPRESSION:
+    case folly::compression::CodecType::NO_COMPRESSION:
       return CompressionKind_NONE;
-    case folly::io::CodecType::ZLIB:
+    case folly::compression::CodecType::ZLIB:
       return CompressionKind_ZLIB;
-    case folly::io::CodecType::SNAPPY:
+    case folly::compression::CodecType::SNAPPY:
       return CompressionKind_SNAPPY;
-    case folly::io::CodecType::ZSTD:
+    case folly::compression::CodecType::ZSTD:
       return CompressionKind_ZSTD;
-    case folly::io::CodecType::LZ4:
+    case folly::compression::CodecType::LZ4:
       return CompressionKind_LZ4;
-    case folly::io::CodecType::GZIP:
+    case folly::compression::CodecType::GZIP:
       return CompressionKind_GZIP;
     default:
       VELOX_UNSUPPORTED(

velox/common/compression/Compression.h

@@ -36,7 +36,7 @@ enum CompressionKind {
 std::unique_ptr<folly::compression::Codec> compressionKindToCodec(
     CompressionKind kind);
 
-CompressionKind codecTypeToCompressionKind(folly::io::CodecType type);
+CompressionKind codecTypeToCompressionKind(folly::compression::CodecType type);
 
 /// Get the name of the CompressionKind.
 std::string compressionKindToString(CompressionKind kind);

velox/common/compression/tests/CompressionTest.cpp

@@ -38,10 +38,10 @@ TEST_F(CompressionTest, testCompressionNames) {
 
 TEST_F(CompressionTest, compressionKindToCodec) {
   ASSERT_EQ(
-      folly::io::CodecType::NO_COMPRESSION,
+      folly::compression::CodecType::NO_COMPRESSION,
       compressionKindToCodec(CompressionKind::CompressionKind_NONE)->type());
   ASSERT_EQ(
-      folly::io::CodecType::LZ4,
+      folly::compression::CodecType::LZ4,
       compressionKindToCodec(CompressionKind::CompressionKind_LZ4)->type());
   EXPECT_THROW(
       compressionKindToCodec(CompressionKind::CompressionKind_LZO),

velox/common/config/GlobalConfig.h

@@ -77,6 +77,14 @@ struct GlobalConfiguration {
   /// and be writable. This flag is ignored if
   /// saveInputOnExpressionAnyFailurePath flag is set.
   std::string saveInputOnExpressionSystemFailurePath;
+  /// Use O_DIRECT for SSD cache I/O. This allows to bypass Linux Kernel's page
+  /// cache and can improve performance on some filesystems. Disable if the
+  /// filesystem does not support it.
+  bool useSsdODirect{true};
+  /// Verify the data written to SSD. Once an entry is written, it is
+  /// immediately read back and is compared against the entry written. This is
+  /// helpful to protect against SSD write corruption.
+  bool verifySsdWrite{false};
 };
 
 GlobalConfiguration& globalConfig();

velox/common/encode/Coding.h

@@ -264,7 +264,7 @@ class Varint {
 // Zig-zag encoding that maps signed integers with a small absolute value
 // to unsigned integers with a small (positive) value.
 // if x >= 0, ZigZag::encode(x) == 2*x
-// if x <  0, ZigZag::encode(x) == -2*x + 1
+// if x < 0, ZigZag::encode(x) == -2*x + 1
 class ZigZag {
  public:
   static uint64_t encode(int64_t val) {
@@ -273,6 +273,10 @@ class ZigZag {
     return (static_cast<uint64_t>(val) << 1) ^ (val >> 63);
   }
 
+  static __uint128_t encodeInt128(__int128_t val) {
+    return (static_cast<__uint128_t>(val) << 1) ^ (val >> 127);
+  }
+
   template <typename U, typename T = typename std::make_signed<U>::type>
   static T decode(U val) {
     return static_cast<T>((val >> 1) ^ -(val & 1));

velox/common/encode/tests/CMakeLists.txt

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-add_executable(velox_common_encode_test Base64Test.cpp)
+add_executable(velox_common_encode_test Base64Test.cpp ZigZagTest.cpp)
 add_test(velox_common_encode_test velox_common_encode_test)
 target_link_libraries(
   velox_common_encode_test

velox/common/encode/tests/ZigZagTest.cpp

@@ -0,0 +1,41 @@
+/*
+ * 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 <gtest/gtest.h>
+#include "velox/common/base/tests/GTestUtils.h"
+#include "velox/common/encode/Coding.h"
+#include "velox/type/HugeInt.h"
+
+namespace facebook::velox::encode::test {
+
+class ZigZagTest : public ::testing::Test {};
+
+TEST_F(ZigZagTest, hugeInt) {
+  auto assertZigZag = [](int128_t value) {
+    auto encoded = ZigZag::encodeInt128(value);
+    auto decoded = ZigZag::decode(encoded);
+    EXPECT_EQ(value, decoded);
+  };
+
+  assertZigZag(0);
+  assertZigZag(HugeInt::parse("1234567890123456789"));
+  assertZigZag(HugeInt::parse("-1234567890123456789"));
+  assertZigZag(HugeInt::parse(std::string(38, '9')));
+  assertZigZag(std::numeric_limits<__int128_t>::max());
+  assertZigZag(std::numeric_limits<__int128_t>::min());
+}
+
+} // namespace facebook::velox::encode::test

velox/common/fuzzer/tests/ConstrainedGeneratorsTest.cpp

@@ -18,6 +18,7 @@
 
 #include <gtest/gtest.h>
 
+#include "velox/common/memory/Memory.h"
 #include "velox/functions/prestosql/types/JsonType.h"
 #include "velox/type/Variant.h"