Merge branch 'facebookincubator:main' into simoneves/15793_support_decimal_types_in_cudf_gpu_operators

Author: simoneves

velox/common/base/tests/GTestUtils.h

@@ -117,6 +117,22 @@
     }                                                                   \
   }
 
+#define VELOX_ASSERT_EQ_TYPES(actual, expected)                         \
+  {                                                                     \
+    auto _actualType = (actual);                                        \
+    auto _expectedType = (expected);                                    \
+    if (_expectedType != nullptr) {                                     \
+      ASSERT_TRUE(_actualType != nullptr)                               \
+          << "Expected: " << _expectedType->toString() << ", got null"; \
+      ASSERT_EQ(*_actualType, *_expectedType)                           \
+          << "Expected: " << _expectedType->toString() << ", got "      \
+          << _actualType->toString();                                   \
+    } else {                                                            \
+      ASSERT_EQ(_actualType, nullptr)                                   \
+          << "Expected null, got " << _actualType->toString();          \
+    }                                                                   \
+  }
+
 #ifndef NDEBUG
 #define DEBUG_ONLY_TEST(test_fixture, test_name) TEST(test_fixture, test_name)
 #define DEBUG_ONLY_TEST_F(test_fixture, test_name) \

velox/connectors/hive/BufferedInputBuilder.cpp

@@ -0,0 +1,47 @@
+/*
+ * 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/connectors/hive/BufferedInputBuilder.h"
+#include "velox/connectors/hive/HiveConnectorUtil.h"
+
+namespace facebook::velox::connector::hive {
+
+class DefaultBufferInputBuilder : public BufferedInputBuilder {
+ public:
+  std::unique_ptr<dwio::common::BufferedInput> create(
+      const FileHandle& fileHandle,
+      const dwio::common::ReaderOptions& readerOpts,
+      const ConnectorQueryCtx* connectorQueryCtx,
+      std::shared_ptr<io::IoStatistics> ioStats,
+      std::shared_ptr<filesystems::File::IoStats> fsStats,
+      folly::Executor* executor,
+      const folly::F14FastMap<std::string, std::string>& fileReadOps) override {
+    return createBufferedInput(
+        fileHandle,
+        readerOpts,
+        connectorQueryCtx,
+        ioStats,
+        fsStats,
+        executor,
+        fileReadOps);
+  }
+};
+
+// static
+std::shared_ptr<BufferedInputBuilder> BufferedInputBuilder::builder_ =
+    std::make_shared<DefaultBufferInputBuilder>();
+
+} // namespace facebook::velox::connector::hive

velox/connectors/hive/BufferedInputBuilder.h

@@ -0,0 +1,56 @@
+/*
+ * 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 <folly/Executor.h>
+
+#include "velox/connectors/Connector.h"
+#include "velox/connectors/hive/FileHandle.h"
+#include "velox/dwio/common/BufferedInput.h"
+#include "velox/dwio/common/Reader.h"
+
+namespace facebook::velox::connector::hive {
+
+/// Registering a different implementation of BufferedInput is allowed using
+/// 'registerBuilder' API.
+class BufferedInputBuilder {
+ public:
+  virtual ~BufferedInputBuilder() = default;
+
+  static const std::shared_ptr<BufferedInputBuilder>& getInstance() {
+    VELOX_CHECK_NOT_NULL(builder_, "Builder is not registered");
+    return builder_;
+  }
+
+  static void registerBuilder(std::shared_ptr<BufferedInputBuilder> builder) {
+    VELOX_CHECK_NOT_NULL(builder);
+    builder_ = std::move(builder);
+  }
+
+  virtual std::unique_ptr<dwio::common::BufferedInput> create(
+      const FileHandle& fileHandle,
+      const dwio::common::ReaderOptions& readerOpts,
+      const ConnectorQueryCtx* connectorQueryCtx,
+      std::shared_ptr<io::IoStatistics> ioStats,
+      std::shared_ptr<filesystems::File::IoStats> fsStats,
+      folly::Executor* executor,
+      const folly::F14FastMap<std::string, std::string>& fileReadOps = {}) = 0;
+
+ private:
+  static std::shared_ptr<BufferedInputBuilder> builder_;
+};
+
+} // namespace facebook::velox::connector::hive

velox/connectors/hive/CMakeLists.txt

@@ -20,6 +20,7 @@ add_subdirectory(iceberg)
 velox_add_library(
   velox_hive_connector
   OBJECT
+  BufferedInputBuilder.cpp
   FileHandle.cpp
   HiveConfig.cpp
   HiveConnector.cpp

velox/connectors/hive/SplitReader.cpp

@@ -17,6 +17,7 @@
 #include "velox/connectors/hive/SplitReader.h"
 
 #include "velox/common/caching/CacheTTLController.h"
+#include "velox/connectors/hive/BufferedInputBuilder.h"
 #include "velox/connectors/hive/HiveConfig.h"
 #include "velox/connectors/hive/HiveConnectorSplit.h"
 #include "velox/connectors/hive/HiveConnectorUtil.h"
@@ -330,7 +331,7 @@ void SplitReader::createReader(
   if (auto* cacheTTLController = cache::CacheTTLController::getInstance()) {
     cacheTTLController->addOpenFileInfo(fileHandleCachePtr->uuid.id());
   }
-  auto baseFileInput = createBufferedInput(
+  auto baseFileInput = BufferedInputBuilder::getInstance()->create(
       *fileHandleCachePtr,
       baseReaderOpts_,
       connectorQueryCtx_,

velox/connectors/hive/iceberg/PositionalDeleteFileReader.cpp

@@ -16,6 +16,7 @@
 
 #include "velox/connectors/hive/iceberg/PositionalDeleteFileReader.h"
 
+#include "velox/connectors/hive/BufferedInputBuilder.h"
 #include "velox/connectors/hive/HiveConnectorUtil.h"
 #include "velox/connectors/hive/TableHandle.h"
 #include "velox/connectors/hive/iceberg/IcebergDeleteFile.h"
@@ -97,7 +98,7 @@ PositionalDeleteFileReader::PositionalDeleteFileReader(
       .filename = deleteFile_.filePath,
       .tokenProvider = connectorQueryCtx_->fsTokenProvider()};
   auto deleteFileHandleCachePtr = fileHandleFactory_->generate(fileHandleKey);
-  auto deleteFileInput = createBufferedInput(
+  auto deleteFileInput = BufferedInputBuilder::getInstance()->create(
       *deleteFileHandleCachePtr,
       deleteReaderOpts,
       connectorQueryCtx,

velox/duckdb/conversion/DuckConversion.cpp

@@ -192,6 +192,9 @@ TypePtr toVeloxType(LogicalType type, bool fileColumnNamesReadAsLowerCase) {
       if (auto customType = getCustomType(name, {})) {
         return customType;
       }
+      if (name == "OPAQUE<void>") {
+        return OPAQUE<void>();
+      }
       [[fallthrough]];
     }
     default:

velox/dwio/common/tests/TestBufferedInput.cpp

@@ -16,6 +16,8 @@
 
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
+#include "velox/common/base/tests/GTestUtils.h"
+#include "velox/connectors/hive/BufferedInputBuilder.h"
 #include "velox/dwio/common/BufferedInput.h"
 
 using namespace facebook::velox::dwio::common;
@@ -144,7 +146,6 @@ class TestBufferedInput : public testing::Test {
 
   const std::shared_ptr<MemoryPool> pool_ = memoryManager()->addLeafPool();
 };
-} // namespace
 
 TEST_F(TestBufferedInput, ZeroLengthStream) {
   auto readFile =
@@ -391,3 +392,45 @@ TEST_F(TestBufferedInput, VReadSortingWithLabels) {
     EXPECT_EQ(next.value(), r.second);
   }
 }
+
+class CustomBufferedInputBuilder
+    : public facebook::velox::connector::hive::BufferedInputBuilder {
+ public:
+  std::unique_ptr<facebook::velox::dwio::common::BufferedInput> create(
+      const facebook::velox::FileHandle& fileHandle,
+      const facebook::velox::dwio::common::ReaderOptions& readerOpts,
+      const facebook::velox::connector::ConnectorQueryCtx* connectorQueryCtx,
+      std::shared_ptr<facebook::velox::io::IoStatistics> ioStats,
+      std::shared_ptr<facebook::velox::filesystems::File::IoStats> fsStats,
+      folly::Executor* executor,
+      const folly::F14FastMap<std::string, std::string>& fileReadOps = {})
+      override {
+    VELOX_NYI("Not implemented in CustomBufferedInputBuilder");
+  }
+};
+
+class CustomBufferedInputTest : public testing::Test {
+ protected:
+  static void SetUpTestCase() {
+    MemoryManager::testingSetInstance(MemoryManager::Options{});
+    facebook::velox::connector::hive::BufferedInputBuilder::registerBuilder(
+        std::make_shared<CustomBufferedInputBuilder>());
+  }
+
+  const std::shared_ptr<MemoryPool> pool_ = memoryManager()->addLeafPool();
+};
+
+} // namespace
+
+TEST_F(CustomBufferedInputTest, basic) {
+  facebook::velox::FileHandle fileHandle;
+  facebook::velox::dwio::common::ReaderOptions readerOpts(pool_.get());
+  auto ioStats = std::make_shared<facebook::velox::io::IoStatistics>();
+  auto fsStats =
+      std::make_shared<facebook::velox::filesystems::File::IoStats>();
+
+  VELOX_ASSERT_THROW(
+      facebook::velox::connector::hive::BufferedInputBuilder::getInstance()
+          ->create(fileHandle, readerOpts, nullptr, ioStats, fsStats, nullptr),
+      "Not implemented in CustomBufferedInputBuilder");
+}