Introducing IcebergSplitReader

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

Add subfield filter for the delete file path column

velox/connectors/hive/CMakeLists.txt

@@ -13,9 +13,10 @@
 # limitations under the License.
 
 add_library(velox_hive_config OBJECT HiveConfig.cpp)
-
 target_link_libraries(velox_hive_config velox_exception)
 
+add_subdirectory(iceberg)
+
 add_library(
   velox_hive_connector OBJECT
   FileHandle.cpp
@@ -31,18 +32,19 @@ add_library(
 
 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)
+  PUBLIC velox_hive_iceberg_splitreader
+  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)
 
 add_library(velox_hive_partition_function HivePartitionFunction.cpp)
 

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/HiveDataSource.cpp

@@ -105,8 +105,8 @@ HiveDataSource::HiveDataSource(
       outputType_(outputType),
       expressionEvaluator_(connectorQueryCtx->expressionEvaluator()),
       fileHandleFactory_(fileHandleFactory),
-      connectorQueryCtx_(connectorQueryCtx),
       executor_(executor),
+      connectorQueryCtx_(connectorQueryCtx),
       hiveConfig_(hiveConfig) {
   // Column handled keyed on the column alias, the name used in the query.
   for (const auto& [canonicalizedName, columnHandle] : columnHandles) {
@@ -241,7 +241,6 @@ void HiveDataSource::addSplit(std::shared_ptr<ConnectorSplit> split) {
   if (splitReader_) {
     splitReader_.reset();
   }
-
   splitReader_ = createSplitReader();
   splitReader_->prepareSplit(metadataFilter_, runtimeStats_);
 }

velox/connectors/hive/SplitReader.cpp

@@ -19,6 +19,7 @@
 #include "velox/connectors/hive/HiveConnectorSplit.h"
 #include "velox/connectors/hive/HiveConnectorUtil.h"
 #include "velox/connectors/hive/TableHandle.h"
+#include "velox/connectors/hive/iceberg/IcebergSplitReader.h"
 #include "velox/dwio/common/CachedBufferedInput.h"
 #include "velox/dwio/common/ReaderFactory.h"
 
@@ -36,17 +37,32 @@ std::unique_ptr<SplitReader> SplitReader::create(
     const ConnectorQueryCtx* connectorQueryCtx,
     const std::shared_ptr<HiveConfig> hiveConfig,
     std::shared_ptr<io::IoStatistics> ioStats) {
-  return std::make_unique<SplitReader>(
-      hiveSplit,
-      hiveTableHandle,
-      scanSpec,
-      readerOutputType,
-      partitionKeys,
-      fileHandleFactory,
-      executor,
-      connectorQueryCtx,
-      hiveConfig,
-      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,
+        executor,
+        connectorQueryCtx,
+        hiveConfig,
+        ioStats);
+  } else {
+    return std::make_unique<SplitReader>(
+        hiveSplit,
+        hiveTableHandle,
+        scanSpec,
+        readerOutputType,
+        partitionKeys,
+        fileHandleFactory,
+        executor,
+        connectorQueryCtx,
+        hiveConfig,
+        ioStats);
+  }
 }
 
 SplitReader::SplitReader(

velox/connectors/hive/SplitReader.h

@@ -38,7 +38,7 @@ namespace facebook::velox::dwio::common {
 class Reader;
 class RowReader;
 class RuntimeStatistics;
-}
+} // namespace facebook::velox::dwio::common
 
 namespace facebook::velox::memory {
 class MemoryPool;

velox/connectors/hive/iceberg/CMakeLists.txt

@@ -0,0 +1,28 @@
+# 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 IcebergSplitReader.cpp IcebergSplit.cpp
+                                 PositionalDeleteFileReader.cpp)
+
+target_link_libraries(
+  velox_hive_iceberg_splitreader
+  Folly::folly
+  gflags::gflags
+  glog::glog
+  gtest
+  gtest_main
+  xsimd)
+
+add_subdirectory(tests)

velox/connectors/hive/iceberg/IcebergDeleteFile.h

@@ -0,0 +1,69 @@
+/*
+ * 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 <string>
+#include <unordered_map>
+#include <vector>
+
+#include "velox/dwio/common/Options.h"
+
+namespace facebook::velox::connector::hive::iceberg {
+
+enum class FileContent {
+  kData,
+  kPositionalDeletes,
+  kEqualityDeletes,
+};
+
+struct IcebergDeleteFile {
+  FileContent content;
+  const std::string filePath;
+  dwio::common::FileFormat fileFormat;
+  uint64_t recordCount;
+  uint64_t fileSizeInBytes;
+  // The field ids for the delete columns for equality delete files
+  std::vector<int32_t> equalityFieldIds;
+  // The lower bounds of the in-file positions for the deleted rows, identified
+  // by each column's field id. E.g. The deleted rows for a column with field id
+  // 1 is in range [10, 50], where 10 and 50 are the deleted row positions in
+  // the data file, then lowerBounds would contain entry <1, "10">
+  std::unordered_map<int32_t, std::string> lowerBounds;
+  // The upper bounds of the in-file positions for the deleted rows, identified
+  // by each column's field id. E.g. The deleted rows for a column with field id
+  // 1 is in range [10, 50], then upperBounds will contain entry <1, "50">
+  std::unordered_map<int32_t, std::string> upperBounds;
+
+  IcebergDeleteFile(
+      FileContent _content,
+      const std::string& _filePath,
+      dwio::common::FileFormat _fileFormat,
+      uint64_t _recordCount,
+      uint64_t _fileSizeInBytes,
+      std::vector<int32_t> _equalityFieldIds = {},
+      std::unordered_map<int32_t, std::string> _lowerBounds = {},
+      std::unordered_map<int32_t, std::string> _upperBounds = {})
+      : content(_content),
+        filePath(_filePath),
+        fileFormat(_fileFormat),
+        recordCount(_recordCount),
+        fileSizeInBytes(_fileSizeInBytes),
+        equalityFieldIds(_equalityFieldIds),
+        lowerBounds(_lowerBounds),
+        upperBounds(_upperBounds) {}
+};
+
+} // namespace facebook::velox::connector::hive::iceberg

velox/connectors/hive/iceberg/IcebergMetadataColumns.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 <string>
+
+#include "velox/type/Type.h"
+
+namespace facebook::velox::connector::hive::iceberg {
+
+const std::string kIcebergDeleteFilePathColumn = "file_path";
+const std::string kIcebergRowPositionColumn = "pos";
+
+struct IcebergMetadataColumn {
+  int id;
+  std::string name;
+  std::shared_ptr<const Type> type;
+  std::string doc;
+
+  IcebergMetadataColumn(
+      int _id,
+      const std::string& _name,
+      std::shared_ptr<const Type> _type,
+      const std::string& _doc)
+      : id(_id), name(_name), type(_type), doc(_doc) {}
+};
+
+#define ICEBERG_DELETE_FILE_PATH_COLUMN()  \
+  std::make_shared<IcebergMetadataColumn>( \
+      2147483546,                          \
+      kIcebergDeleteFilePathColumn,        \
+      VARCHAR(),                           \
+      "Path of a file in which a deleted row is stored")
+
+#define ICEBERG_DELETE_FILE_POSITIONS_COLUMN() \
+  std::make_shared<IcebergMetadataColumn>(     \
+      2147483545,                              \
+      kIcebergRowPositionColumn,               \
+      BIGINT(),                                \
+      "Ordinal position of a deleted row in the data file")
+
+} // namespace facebook::velox::connector::hive::iceberg

velox/connectors/hive/iceberg/IcebergSplit.cpp

@@ -0,0 +1,69 @@
+/*
+ * 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/iceberg/IcebergSplit.h"
+
+#include "velox/connectors/hive/iceberg/IcebergDeleteFile.h"
+
+namespace facebook::velox::connector::hive::iceberg {
+
+HiveIcebergSplit::HiveIcebergSplit(
+    const std::string& _connectorId,
+    const std::string& _filePath,
+    dwio::common::FileFormat _fileFormat,
+    uint64_t _start,
+    uint64_t _length,
+    const std::unordered_map<std::string, std::optional<std::string>>&
+        _partitionKeys,
+    std::optional<int32_t> _tableBucketNumber,
+    const std::unordered_map<std::string, std::string>& _customSplitInfo,
+    const std::shared_ptr<std::string>& _extraFileInfo)
+    : HiveConnectorSplit(
+          _connectorId,
+          _filePath,
+          _fileFormat,
+          _start,
+          _length,
+          _partitionKeys,
+          _tableBucketNumber) {
+  // TODO: Deserialize _extraFileInfo to get deleteFiles;
+}
+
+// For tests only
+HiveIcebergSplit::HiveIcebergSplit(
+    const std::string& _connectorId,
+    const std::string& _filePath,
+    dwio::common::FileFormat _fileFormat,
+    uint64_t _start,
+    uint64_t _length,
+    const std::unordered_map<std::string, std::optional<std::string>>&
+        _partitionKeys,
+    std::optional<int32_t> _tableBucketNumber,
+    const std::unordered_map<std::string, std::string>& _customSplitInfo,
+    const std::shared_ptr<std::string>& _extraFileInfo,
+    std::vector<IcebergDeleteFile> _deletes)
+    : HiveConnectorSplit(
+          _connectorId,
+          _filePath,
+          _fileFormat,
+          _start,
+          _length,
+          _partitionKeys,
+          _tableBucketNumber,
+          _customSplitInfo,
+          _extraFileInfo),
+      deleteFiles(_deletes) {}
+} // namespace facebook::velox::connector::hive::iceberg