Iceberg refactor into a separate lakehouse connector

.github/workflows/gluten_daily.yml

@@ -0,0 +1,75 @@
+# 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.
+
+name: Gluten Daily Build
+
+on:
+  push:
+    branches:
+      - 'main'
+
+concurrency:
+  group: ${{ github.repository }}-${{ github.head_ref || github.sha }}-${{ github.workflow }}
+  cancel-in-progress: true
+
+jobs:
+
+  gluten-cpp-build:
+    name: gluten cpp build
+    # prevent errors when forks ff their main branch
+    if: ${{ github.repository == 'IBM/velox' }}
+    runs-on: ubuntu-22.04
+    env:
+      CCACHE_DIR: "${{ github.workspace }}/.ccache"
+    steps:
+      - uses: actions/checkout@v4
+      - name: Get Ccache
+        uses: actions/cache/restore@v4
+        with:
+          path: '${{ env.CCACHE_DIR }}'
+          key: ccache-centos7-release-default-${{github.sha}}
+          restore-keys: |
+            ccache-centos7-release-default
+      - name: Setup Gluten
+        run: |
+          git clone --depth 1 https://github.com/apache/incubator-gluten gluten && cd gluten
+          BRANCH=$(echo ${GITHUB_REF#refs/heads/})
+          sed -i 's/oap-project/IBM/g' ep/build-velox/src/get_velox.sh
+          #sed -i 's/VELOX_BRANCH=2025.*/VELOX_BRANCH=main/g' ep/build-velox/src/get_velox.sh
+      - name: Build Gluten native libraries
+        run: |
+          docker pull apache/gluten:vcpkg-centos-7
+          docker run -v $GITHUB_WORKSPACE:/work -w /work apache/gluten:vcpkg-centos-7 bash -c "
+            git config --global --add safe.directory /work
+            set -e
+            df -a
+            cd /work/gluten
+            export CCACHE_DIR=/work/.ccache
+            mkdir -p /work/.ccache
+            source /opt/rh/devtoolset-11/enable
+            source /opt/rh/rh-git227/enable
+            export NUM_THREADS=4
+            ccache -sz
+            ./dev/builddeps-veloxbe.sh --enable_vcpkg=ON --build_arrow=OFF --build_tests=OFF --build_benchmarks=OFF \
+                           --build_examples=OFF --enable_s3=ON --enable_gcs=ON --enable_hdfs=ON --enable_abfs=ON --velox_home=/work
+            ccache -s
+            mkdir -p /work/.m2/repository/org/apache/arrow/
+            cp -r /root/.m2/repository/org/apache/arrow/* /work/.m2/repository/org/apache/arrow/
+          "
+      - name: Save ccache
+        uses: actions/cache/save@v4
+        id: ccache
+        with:
+          path: '${{ env.CCACHE_DIR }}'
+          key: ccache-centos7-release-default-${{github.sha}}

.github/workflows/linux-build.yml

@@ -15,21 +15,6 @@
 name: Linux Build using GCC
 
 on:
-  push:
-    branches:
-      - main
-    paths:
-      - velox/**
-      - '!velox/docs/**'
-      - CMakeLists.txt
-      - CMake/**
-      - scripts/setup-ubuntu.sh
-      - scripts/setup-common.sh
-      - scripts/setup-versions.sh
-      - scripts/setup-helper-functions.sh
-      - .github/workflows/linux-build.yml
-      - .github/workflows/linux-build-base.yml
-
   pull_request:
     paths:
       - velox/**

CMakeLists.txt

@@ -136,7 +136,9 @@ set(
 )
 option(VELOX_ENABLE_EXEC "Build exec." ON)
 option(VELOX_ENABLE_AGGREGATES "Build aggregates." ON)
-option(VELOX_ENABLE_HIVE_CONNECTOR "Build Hive connector." ON)
+option(VELOX_ENABLE_HIVE_CONNECTOR "Build the Hive connector." ON)
+#option(VELOX_ENABLE_HIVE_NEW_CONNECTOR "Build the new Hive connector." ON)
+#option(VELOX_ENABLE_ICEBERG_CONNECTOR "Build the ICEBERG connector that does NOT depend on the new Hive connector." ON)
 option(VELOX_ENABLE_TPCH_CONNECTOR "Build TPC-H connector." ON)
 option(VELOX_ENABLE_TPCDS_CONNECTOR "Build TPC-DS connector." ON)
 option(VELOX_ENABLE_PRESTO_FUNCTIONS "Build Presto SQL functions." ON)
@@ -704,6 +706,7 @@ include_directories(.)
 
 # Adding this down here prevents warnings in dependencies from stopping the
 # build
+set(TREAT_WARNINGS_AS_ERRORS OFF)
 if("${TREAT_WARNINGS_AS_ERRORS}")
   set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Werror")
 endif()

velox/common/memory/tests/SharedArbitratorTest.cpp

@@ -755,7 +755,7 @@ DEBUG_ONLY_TEST_P(
     folly::EventCount taskPauseWait;
     auto taskPauseWaitKey = taskPauseWait.prepareWait();
 
-    const auto fakeAllocationSize = kMemoryCapacity - (32L << 20);
+    const auto fakeAllocationSize = kMemoryCapacity - (2L << 20);
 
     std::atomic<bool> injectAllocationOnce{true};
     fakeOperatorFactory_->setAllocationCallback([&](Operator* op) {

velox/connectors/CMakeLists.txt

@@ -17,6 +17,8 @@ velox_link_libraries(velox_connector velox_common_config velox_vector)
 
 add_subdirectory(fuzzer)
 
+add_subdirectory(lakehouse)
+
 if(${VELOX_ENABLE_HIVE_CONNECTOR})
   add_subdirectory(hive)
 endif()

velox/connectors/Connector.h

@@ -191,7 +191,7 @@ class DataSink {
     uint64_t recodeTimeNs{0};
     uint64_t compressionTimeNs{0};
 
-    common::SpillStats spillStats;
+    velox::common::SpillStats spillStats;
 
     bool empty() const;
 

velox/connectors/hive/CMakeLists.txt

@@ -15,8 +15,6 @@
 velox_add_library(velox_hive_config OBJECT HiveConfig.cpp)
 velox_link_libraries(velox_hive_config velox_core velox_exception)
 
-add_subdirectory(iceberg)
-
 velox_add_library(
   velox_hive_connector
   OBJECT
@@ -35,7 +33,6 @@ velox_add_library(
 
 velox_link_libraries(
   velox_hive_connector
-  PUBLIC velox_hive_iceberg_splitreader
   PRIVATE velox_common_io velox_connector velox_dwio_catalog_fbhive velox_hive_partition_function
 )
 

velox/connectors/hive/HiveConfig.cpp

@@ -273,4 +273,9 @@ std::string HiveConfig::schema(const config::ConfigBase* session) const {
       kSchema, config_->get<std::string>(kSchema, ""));
 }
 
+bool HiveConfig::fanoutEnabled(const config::ConfigBase* session) const {
+  return session->get<bool>(
+      kFanoutEnabledSession, config_->get<bool>(kFanoutEnabled, true));
+}
+
 } // namespace facebook::velox::connector::hive

velox/connectors/hive/HiveConfig.h

@@ -210,6 +210,12 @@ class HiveConfig {
   static constexpr const char* kSource = "source";
   static constexpr const char* kSchema = "schema";
 
+  /// Controls the writer mode, whether the fanout mode writer is enabled,
+  /// default value is true, setting to false means clustered mode.
+  /// Currently applies only to the Iceberg writer.
+  static constexpr const char* kFanoutEnabled = "fanout-enabled";
+  static constexpr const char* kFanoutEnabledSession = "fanout_enabled";
+
   InsertExistingPartitionsBehavior insertExistingPartitionsBehavior(
       const config::ConfigBase* session) const;
 
@@ -307,6 +313,9 @@ class HiveConfig {
   /// Schema of the query. Used for storage logging.
   std::string schema(const config::ConfigBase* session) const;
 
+  /// Return if fanout writer mode is enabled.
+  bool fanoutEnabled(const config::ConfigBase* session) const;
+
   HiveConfig(std::shared_ptr<const config::ConfigBase> config) {
     VELOX_CHECK_NOT_NULL(
         config, "Config is null for HiveConfig initialization");

velox/connectors/hive/HiveConnector.cpp

@@ -20,7 +20,6 @@
 #include "velox/connectors/hive/HiveDataSink.h"
 #include "velox/connectors/hive/HiveDataSource.h"
 #include "velox/connectors/hive/HivePartitionFunction.h"
-#include "velox/connectors/hive/iceberg/IcebergDataSink.h"
 
 #include <boost/lexical_cast.hpp>
 #include <memory>
@@ -74,16 +73,6 @@ std::unique_ptr<DataSink> HiveConnector::createDataSink(
     ConnectorInsertTableHandlePtr connectorInsertTableHandle,
     ConnectorQueryCtx* connectorQueryCtx,
     CommitStrategy commitStrategy) {
-  if (auto icebergInsertHandle =
-          std::dynamic_pointer_cast<const iceberg::IcebergInsertTableHandle>(
-              connectorInsertTableHandle)) {
-    return std::make_unique<iceberg::IcebergDataSink>(
-        inputType,
-        icebergInsertHandle,
-        connectorQueryCtx,
-        commitStrategy,
-        hiveConfig_);
-  }
   auto hiveInsertHandle =
       std::dynamic_pointer_cast<const HiveInsertTableHandle>(
           connectorInsertTableHandle);

velox/connectors/hive/HiveConnectorUtil.cpp

@@ -24,6 +24,10 @@
 #include "velox/expression/ExprConstants.h"
 #include "velox/expression/ExprToSubfieldFilter.h"
 
+#include <boost/lexical_cast.hpp>
+#include <boost/uuid/uuid_generators.hpp>
+#include <boost/uuid/uuid_io.hpp>
+
 namespace facebook::velox::connector::hive {
 namespace {
 
@@ -944,4 +948,9 @@ core::TypedExprPtr extractFiltersFromRemainingFilter(
   }
   return expr;
 }
+
+std::string makeUuid() {
+  return boost::lexical_cast<std::string>(boost::uuids::random_generator()());
+}
+
 } // namespace facebook::velox::connector::hive

velox/connectors/hive/HiveConnectorUtil.h

@@ -116,4 +116,6 @@ core::TypedExprPtr extractFiltersFromRemainingFilter(
     common::SubfieldFilters& filters,
     double& sampleRate);
 
+std::string makeUuid();
+
 } // namespace facebook::velox::connector::hive