fix iceberg reader

Signed-off-by: Yuan <yuanzhou@apache.org>

Author: zhouyuan

velox/connectors/hive/HiveDataSource.cpp

@@ -137,7 +137,7 @@ std::unique_ptr<FileSplitReader> HiveDataSource::createSplitReader() {
   auto bucketChannels = prepareSplit();
   auto hiveSplit = checkedPointerCast<const HiveConnectorSplit>(split_);
 
-  return std::make_unique<HiveSplitReader>(
+  return HiveSplitReader::create(
       hiveSplit,
       tableHandle_,
       &partitionKeys_,

velox/connectors/hive/HiveSplitReader.cpp

@@ -19,9 +19,64 @@
 #include "velox/connectors/hive/FileConfig.h"
 #include "velox/connectors/hive/HiveConnectorSplit.h"
 #include "velox/connectors/hive/HiveConnectorUtil.h"
+#include "velox/connectors/hive/iceberg/IcebergSplit.h"
+#include "velox/connectors/hive/iceberg/IcebergSplitReader.h"
 
 namespace facebook::velox::connector::hive {
 
+std::unique_ptr<FileSplitReader> HiveSplitReader::create(
+    const std::shared_ptr<const HiveConnectorSplit>& hiveSplit,
+    const FileTableHandlePtr& tableHandle,
+    const std::unordered_map<std::string, FileColumnHandlePtr>* partitionKeys,
+    const ConnectorQueryCtx* connectorQueryCtx,
+    const std::shared_ptr<const FileConfig>& fileConfig,
+    const RowTypePtr& readerOutputType,
+    const std::shared_ptr<io::IoStatistics>& ioStatistics,
+    const std::shared_ptr<IoStats>& ioStats,
+    FileHandleFactory* fileHandleFactory,
+    folly::Executor* ioExecutor,
+    const std::shared_ptr<common::ScanSpec>& scanSpec,
+    const std::unordered_map<std::string, FileColumnHandlePtr>* infoColumns,
+    std::vector<column_index_t> bucketChannels,
+    const common::SubfieldFilters* subfieldFiltersForValidation) {
+  // Create the SplitReader based on hiveSplit->customSplitInfo["table_format"]
+  if (hiveSplit->customSplitInfo.count("table_format") > 0 &&
+      hiveSplit->customSplitInfo.at("table_format") == "hive-iceberg") {
+    auto icebergSplit =
+        std::dynamic_pointer_cast<const iceberg::HiveIcebergSplit>(hiveSplit);
+    VELOX_CHECK_NOT_NULL(
+        icebergSplit,
+        "Expected HiveIcebergSplit for table_format=hive-iceberg");
+    return std::make_unique<iceberg::IcebergSplitReader>(
+        icebergSplit,
+        tableHandle,
+        partitionKeys,
+        connectorQueryCtx,
+        fileConfig,
+        readerOutputType,
+        ioStatistics,
+        ioStats,
+        fileHandleFactory,
+        ioExecutor,
+        scanSpec);
+  } else {
+    return std::make_unique<HiveSplitReader>(
+        hiveSplit,
+        tableHandle,
+        partitionKeys,
+        connectorQueryCtx,
+        fileConfig,
+        readerOutputType,
+        ioStatistics,
+        ioStats,
+        fileHandleFactory,
+        ioExecutor,
+        scanSpec,
+	      infoColumns,
+        std::move(bucketChannels));
+  }
+}
+
 HiveSplitReader::HiveSplitReader(
     const std::shared_ptr<const HiveConnectorSplit>& hiveSplit,
     const FileTableHandlePtr& tableHandle,

velox/connectors/hive/HiveSplitReader.h

@@ -30,6 +30,25 @@ struct HiveConnectorSplit;
 /// to keep only rows belonging to the target bucket.
 class HiveSplitReader : public FileSplitReader {
  public:
+  /// Factory method to create the appropriate split reader based on split type.
+  /// For Iceberg splits (identified by customSplitInfo["table_format"] == "hive-iceberg"),
+  /// creates an IcebergSplitReader. Otherwise, creates a HiveSplitReader.
+  static std::unique_ptr<FileSplitReader> create(
+      const std::shared_ptr<const HiveConnectorSplit>& hiveSplit,
+      const FileTableHandlePtr& tableHandle,
+      const std::unordered_map<std::string, FileColumnHandlePtr>* partitionKeys,
+      const ConnectorQueryCtx* connectorQueryCtx,
+      const std::shared_ptr<const FileConfig>& fileConfig,
+      const RowTypePtr& readerOutputType,
+      const std::shared_ptr<io::IoStatistics>& ioStatistics,
+      const std::shared_ptr<IoStats>& ioStats,
+      FileHandleFactory* fileHandleFactory,
+      folly::Executor* ioExecutor,
+      const std::shared_ptr<common::ScanSpec>& scanSpec,
+      const std::unordered_map<std::string, FileColumnHandlePtr>* infoColumns,
+      std::vector<column_index_t> bucketChannels = {},
+      const common::SubfieldFilters* subfieldFiltersForValidation = nullptr);
+
   HiveSplitReader(
       const std::shared_ptr<const HiveConnectorSplit>& hiveSplit,
       const FileTableHandlePtr& tableHandle,