Introducing IcebergSplitReader

In this commit we introduces IcebergSplitReader which supports reading
Iceberg splits with positional delete files.

velox/benchmarks/tpch/CMakeLists.txt

@@ -24,6 +24,7 @@ target_link_libraries(
   velox_dwio_parquet_reader
   velox_dwio_common_test_utils
   velox_hive_connector
+  # velox_hive_iceberg_splitreader
   velox_exception
   velox_memory
   velox_process

velox/common/testutil/tests/CMakeLists.txt

@@ -16,11 +16,5 @@ add_executable(velox_test_util_test TestValueTest.cpp SpillConfigTest.cpp)
 gtest_add_tests(velox_test_util_test "" AUTO)
 
 target_link_libraries(
-  velox_test_util_test
-  PRIVATE
-  velox_test_util
-  velox_exception
-  velox_spill_config
-  velox_exec
-  gtest
-  gtest_main)
+  velox_test_util_test PRIVATE velox_test_util velox_exception
+                               velox_spill_config velox_exec gtest gtest_main)

velox/connectors/hive/CMakeLists.txt

@@ -16,6 +16,10 @@ add_library(velox_hive_config OBJECT HiveConfig.cpp)
 
 target_link_libraries(velox_hive_config velox_exception)
 
+add_subdirectory(iceberg)
+
+# find_package(iceberg CONFIG REQUIRED)
+
 add_library(
   velox_hive_connector OBJECT
   FileHandle.cpp
@@ -28,25 +32,28 @@ add_library(
   SplitReader.cpp
   TableHandle.cpp)
 
+# target_include_directories(velox_hive_connector PUBLIC iceberg::iceberg)
 target_link_libraries(
   velox_hive_connector
-  velox_common_io
-  velox_connector
-  velox_dwio_catalog_fbhive
-  velox_dwio_dwrf_reader
-  velox_dwio_dwrf_writer
-  velox_dwio_parquet_reader
-  velox_dwio_parquet_writer
-  velox_file
-  velox_hive_partition_function
-  velox_s3fs
-  velox_hdfs
-  velox_gcs)
+  PRIVATE velox_common_io
+          velox_connector
+          velox_dwio_catalog_fbhive
+          velox_dwio_dwrf_reader
+          velox_dwio_dwrf_writer
+          velox_dwio_parquet_reader
+          velox_dwio_parquet_writer
+          velox_file
+          velox_hive_partition_function
+          velox_s3fs
+          velox_hdfs
+          velox_gcs
+  PUBLIC velox_hive_iceberg_splitreader)
 
 add_library(velox_hive_partition_function HivePartitionFunction.cpp)
 
 target_link_libraries(velox_hive_partition_function velox_core velox_exec)
 
+# add_subdirectory(iceberg)
 add_subdirectory(storage_adapters)
 
 if(${VELOX_BUILD_TESTING})

velox/connectors/hive/FileHandle.h

@@ -25,15 +25,15 @@
 
 #pragma once
 
-#include <cstdint>
-#include <memory>
-#include <string>
-
 #include "velox/common/caching/CachedFactory.h"
 #include "velox/common/caching/FileIds.h"
 #include "velox/common/file/File.h"
 #include "velox/dwio/common/InputStream.h"
 
+//#include <cstdint>
+//#include <memory>
+//#include <string>
+
 namespace facebook::velox {
 
 class Config;

velox/connectors/hive/HiveConnector.cpp

@@ -75,7 +75,6 @@ std::unique_ptr<DataSource> HiveConnector::createDataSource(
         std::string,
         std::shared_ptr<connector::ColumnHandle>>& columnHandles,
     ConnectorQueryCtx* connectorQueryCtx) {
-
   return std::make_unique<HiveDataSource>(
       outputType,
       tableHandle,

velox/connectors/hive/HiveDataSource.cpp

@@ -501,8 +501,8 @@ std::unique_ptr<SplitReader> HiveDataSource::createSplitReader() {
       readerOutputType_,
       &partitionKeys_,
       fileHandleFactory_,
-      executor_,
       connectorQueryCtx_,
+      executor_,
       ioStats_);
 }
 
@@ -518,7 +518,6 @@ void HiveDataSource::addSplit(std::shared_ptr<ConnectorSplit> split) {
   if (splitReader_) {
     splitReader_.reset();
   }
-
   splitReader_ = createSplitReader();
   splitReader_->prepareSplit(metadataFilter_, runtimeStats_);
 }

velox/connectors/hive/HiveDataSource.h

@@ -141,9 +141,9 @@ class HiveDataSource : public DataSource {
   SelectivityVector filterRows_;
   exec::FilterEvalCtx filterEvalCtx_;
 
-  FileHandleFactory* fileHandleFactory_;
-  const ConnectorQueryCtx* const connectorQueryCtx_;
-  folly::Executor* executor_;
+  FileHandleFactory* const fileHandleFactory_;
+  ConnectorQueryCtx* const connectorQueryCtx_;
+  folly::Executor* const executor_;
 };
 
 } // namespace facebook::velox::connector::hive

velox/connectors/hive/SplitReader.cpp

@@ -19,12 +19,13 @@
 #include "velox/connectors/hive/HiveConfig.h"
 #include "velox/connectors/hive/HiveConnectorSplit.h"
 #include "velox/connectors/hive/TableHandle.h"
+#include "velox/connectors/hive/iceberg/IcebergSplitReader.h"
 #include "velox/dwio/common/CachedBufferedInput.h"
 #include "velox/dwio/common/Options.h"
 #include "velox/dwio/common/ReaderFactory.h"
 
 #include <folly/Conv.h>
-#include <gtest/gtest.h>
+// #include <gtest/gtest.h>
 
 #include <string>
 #include <unordered_map>
@@ -144,19 +145,34 @@ std::unique_ptr<SplitReader> SplitReader::create(
     std::unordered_map<std::string, std::shared_ptr<HiveColumnHandle>>*
         partitionKeys,
     FileHandleFactory* fileHandleFactory,
+    ConnectorQueryCtx* connectorQueryCtx,
     folly::Executor* executor,
-    const ConnectorQueryCtx* connectorQueryCtx,
     std::shared_ptr<io::IoStatistics> ioStats) {
-  return std::make_unique<SplitReader>(
-      hiveSplit,
-      hiveTableHandle,
-      scanSpec,
-      readerOutputType,
-      partitionKeys,
-      fileHandleFactory,
-      executor,
-      connectorQueryCtx,
-      ioStats);
+  //  Create the SplitReader based on hiveSplit->customSplitInfo["table_format"]
+  if (hiveSplit->customSplitInfo["table_format"] == "hive_iceberg") {
+    return std::make_unique<iceberg::IcebergSplitReader>(
+        hiveSplit,
+        hiveTableHandle,
+        scanSpec,
+        readerOutputType,
+        partitionKeys,
+        fileHandleFactory,
+
+        connectorQueryCtx,
+        executor,
+        ioStats);
+  } else {
+    return std::make_unique<SplitReader>(
+        hiveSplit,
+        hiveTableHandle,
+        scanSpec,
+        readerOutputType,
+        partitionKeys,
+        fileHandleFactory,
+        connectorQueryCtx,
+        executor,
+        ioStats);
+  }
 }
 
 SplitReader::SplitReader(
@@ -167,17 +183,17 @@ SplitReader::SplitReader(
     std::unordered_map<std::string, std::shared_ptr<HiveColumnHandle>>*
         partitionKeys,
     FileHandleFactory* fileHandleFactory,
+    ConnectorQueryCtx* connectorQueryCtx,
     folly::Executor* executor,
-    const ConnectorQueryCtx* connectorQueryCtx,
     std::shared_ptr<io::IoStatistics> ioStats)
     : hiveSplit_(hiveSplit),
       hiveTableHandle_(hiveTableHandle),
       scanSpec_(scanSpec),
       readerOutputType_(readerOutputType),
       partitionKeys_(partitionKeys),
       fileHandleFactory_(fileHandleFactory),
-      executor_(executor),
       connectorQueryCtx_(connectorQueryCtx),
+      executor_(executor),
       ioStats_(ioStats),
       baseReaderOpts_(connectorQueryCtx->memoryPool()) {}
 

velox/connectors/hive/SplitReader.h

@@ -51,8 +51,8 @@ class SplitReader {
       std::unordered_map<std::string, std::shared_ptr<HiveColumnHandle>>*
           partitionKeys,
       FileHandleFactory* fileHandleFactory,
+      ConnectorQueryCtx* connectorQueryCtx,
       folly::Executor* executor,
-      const ConnectorQueryCtx* connectorQueryCtx,
       std::shared_ptr<io::IoStatistics> ioStats);
 
   SplitReader(
@@ -63,8 +63,8 @@ class SplitReader {
       std::unordered_map<std::string, std::shared_ptr<HiveColumnHandle>>*
           partitionKeys,
       FileHandleFactory* fileHandleFactory,
+      ConnectorQueryCtx* connectorQueryCtx,
       folly::Executor* executor,
-      const ConnectorQueryCtx* connectorQueryCtx,
       std::shared_ptr<io::IoStatistics> ioStats);
 
   virtual ~SplitReader() = default;
@@ -122,8 +122,8 @@ class SplitReader {
   std::unique_ptr<dwio::common::Reader> baseReader_;
   std::unique_ptr<dwio::common::RowReader> baseRowReader_;
   FileHandleFactory* const fileHandleFactory_;
+  ConnectorQueryCtx* const connectorQueryCtx_;
   folly::Executor* const executor_;
-  const ConnectorQueryCtx* const connectorQueryCtx_;
   std::shared_ptr<io::IoStatistics> ioStats_;
 
  private:

velox/connectors/hive/benchmarks/CMakeLists.txt

@@ -27,6 +27,7 @@ target_link_libraries(
   velox_exec
   velox_exec_test_lib
   velox_hive_connector
+  velox_hive_iceberg_splitreader
   velox_hive_partition_function
   velox_memory
   Folly::folly

velox/connectors/hive/iceberg/CMakeLists.txt

@@ -0,0 +1,37 @@
+# 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_library(velox_hive_iceberg_splitreader OBJECT IMPORTED GLOBAL
+# IcebergSplitReader.cpp IcebergSplit.cpp DeleteFileReader.cpp)
+
+add_library(velox_hive_iceberg_splitreader OBJECT
+            IcebergSplitReader.cpp IcebergSplit.cpp DeleteFileReader.cpp)
+
+target_link_libraries(
+  velox_hive_iceberg_splitreader
+  Folly::folly
+  gtest
+  Boost::headers
+  Boost::filesystem
+  gflags::gflags
+  glog::glog
+  gtest
+  gtest_main
+  xsimd)
+
+# install(TARGETS velox_hive_iceberg_splitreader EXPORT iceberg) message (STATUS
+# "Installed velox_hive_iceberg_splitreader") install(EXPORT iceberg FILE
+# icebergConfig.cmake NAMESPACE iceberg:: DESTINATION lib/cmake/iceberg)
+
+add_subdirectory(tests)