Revert "SplitReader refactor (facebookincubator#8995)"

This reverts commit b5ea2d7.
marin-ma · Apr 15, 2024 · 1492034 · 1492034
1 parent 363e3cc
commit 1492034
Show file tree

Hide file tree

Showing 9 changed files with 230 additions and 280 deletions.
diff --git a/velox/connectors/hive/HiveConnectorUtil.cpp b/velox/connectors/hive/HiveConnectorUtil.cpp
@@ -450,10 +450,10 @@ std::unique_ptr<dwio::common::SerDeOptions> parseSerdeParameters(
 
 void configureReaderOptions(
     dwio::common::ReaderOptions& readerOptions,
-    const std::shared_ptr<const HiveConfig>& hiveConfig,
+    const std::shared_ptr<HiveConfig>& hiveConfig,
     const Config* sessionProperties,
-    const std::shared_ptr<const HiveTableHandle>& hiveTableHandle,
-    const std::shared_ptr<const HiveConnectorSplit>& hiveSplit) {
+    const std::shared_ptr<HiveTableHandle>& hiveTableHandle,
+    const std::shared_ptr<HiveConnectorSplit>& hiveSplit) {
   configureReaderOptions(
       readerOptions,
       hiveConfig,
@@ -465,10 +465,10 @@ void configureReaderOptions(
 
 void configureReaderOptions(
     dwio::common::ReaderOptions& readerOptions,
-    const std::shared_ptr<const HiveConfig>& hiveConfig,
+    const std::shared_ptr<HiveConfig>& hiveConfig,
     const Config* sessionProperties,
     const RowTypePtr& fileSchema,
-    const std::shared_ptr<const HiveConnectorSplit>& hiveSplit,
+    const std::shared_ptr<HiveConnectorSplit>& hiveSplit,
     const std::unordered_map<std::string, std::string>& tableParameters) {
   readerOptions.setLoadQuantum(hiveConfig->loadQuantum());
   readerOptions.setMaxCoalesceBytes(hiveConfig->maxCoalescedBytes());
@@ -502,10 +502,10 @@ void configureReaderOptions(
 void configureRowReaderOptions(
     dwio::common::RowReaderOptions& rowReaderOptions,
     const std::unordered_map<std::string, std::string>& tableParameters,
-    const std::shared_ptr<common::ScanSpec>& scanSpec,
-    const std::shared_ptr<common::MetadataFilter>& metadataFilter,
+    std::shared_ptr<common::ScanSpec> scanSpec,
+    std::shared_ptr<common::MetadataFilter> metadataFilter,
     const RowTypePtr& rowType,
-    const std::shared_ptr<const HiveConnectorSplit>& hiveSplit) {
+    std::shared_ptr<HiveConnectorSplit> hiveSplit) {
   auto skipRowsIt =
       tableParameters.find(dwio::common::TableParameter::kSkipHeaderLineCount);
   if (skipRowsIt != tableParameters.end()) {
@@ -569,15 +569,13 @@ bool applyPartitionFilter(
 } // namespace
 
 bool testFilters(
-    const common::ScanSpec* scanSpec,
-    const dwio::common::Reader* reader,
+    common::ScanSpec* scanSpec,
+    dwio::common::Reader* reader,
     const std::string& filePath,
     const std::unordered_map<std::string, std::optional<std::string>>&
         partitionKey,
-    const std::unordered_map<std::string, std::shared_ptr<HiveColumnHandle>>&
+    std::unordered_map<std::string, std::shared_ptr<HiveColumnHandle>>*
         partitionKeysHandle) {
-  VELOX_CHECK_EQ(partitionKey.size(), partitionKeysHandle.size());
-
   auto totalRows = reader->numberOfRows();
   const auto& fileTypeWithId = reader->typeWithId();
   const auto& rowType = reader->rowType();
@@ -588,11 +586,8 @@ bool testFilters(
         // If missing column is partition key.
         auto iter = partitionKey.find(name);
         if (iter != partitionKey.end() && iter->second.has_value()) {
-          auto handlesIter = partitionKeysHandle.find(name);
-          VELOX_CHECK(handlesIter != partitionKeysHandle.end());
-
           return applyPartitionFilter(
-              handlesIter->second->dataType()->kind(),
+              (*partitionKeysHandle)[name]->dataType()->kind(),
               iter->second.value(),
               child->filter());
         }

diff --git a/velox/connectors/hive/HiveConnectorUtil.h b/velox/connectors/hive/HiveConnectorUtil.h
@@ -61,34 +61,34 @@ std::shared_ptr<common::ScanSpec> makeScanSpec(
 
 void configureReaderOptions(
     dwio::common::ReaderOptions& readerOptions,
-    const std::shared_ptr<const HiveConfig>& config,
+    const std::shared_ptr<HiveConfig>& config,
     const Config* sessionProperties,
-    const std::shared_ptr<const HiveTableHandle>& hiveTableHandle,
-    const std::shared_ptr<const HiveConnectorSplit>& hiveSplit);
+    const std::shared_ptr<HiveTableHandle>& hiveTableHandle,
+    const std::shared_ptr<HiveConnectorSplit>& hiveSplit);
 
 void configureReaderOptions(
     dwio::common::ReaderOptions& readerOptions,
-    const std::shared_ptr<const HiveConfig>& hiveConfig,
+    const std::shared_ptr<HiveConfig>& hiveConfig,
     const Config* sessionProperties,
     const RowTypePtr& fileSchema,
-    const std::shared_ptr<const HiveConnectorSplit>& hiveSplit,
+    const std::shared_ptr<HiveConnectorSplit>& hiveSplit,
     const std::unordered_map<std::string, std::string>& tableParameters = {});
 
 void configureRowReaderOptions(
     dwio::common::RowReaderOptions& rowReaderOptions,
     const std::unordered_map<std::string, std::string>& tableParameters,
-    const std::shared_ptr<common::ScanSpec>& scanSpec,
-    const std::shared_ptr<common::MetadataFilter>& metadataFilter,
+    std::shared_ptr<common::ScanSpec> scanSpec,
+    std::shared_ptr<common::MetadataFilter> metadataFilter,
     const RowTypePtr& rowType,
-    const std::shared_ptr<const HiveConnectorSplit>& hiveSplit);
+    std::shared_ptr<HiveConnectorSplit> hiveSplit);
 
 bool testFilters(
-    const common::ScanSpec* scanSpec,
-    const dwio::common::Reader* reader,
+    common::ScanSpec* scanSpec,
+    dwio::common::Reader* reader,
     const std::string& filePath,
     const std::unordered_map<std::string, std::optional<std::string>>&
         partitionKey,
-    const std::unordered_map<std::string, std::shared_ptr<HiveColumnHandle>>&
+    std::unordered_map<std::string, std::shared_ptr<HiveColumnHandle>>*
         partitionKeysHandle);
 
 std::unique_ptr<dwio::common::BufferedInput> createBufferedInput(

diff --git a/velox/connectors/hive/HiveDataSource.cpp b/velox/connectors/hive/HiveDataSource.cpp
@@ -170,14 +170,14 @@ std::unique_ptr<SplitReader> HiveDataSource::createSplitReader() {
   return SplitReader::create(
       split_,
       hiveTableHandle_,
-      &partitionKeys_,
-      connectorQueryCtx_,
-      hiveConfig_,
+      scanSpec_,
       readerOutputType_,
-      ioStats_,
+      &partitionKeys_,
       fileHandleFactory_,
       executor_,
-      scanSpec_);
+      connectorQueryCtx_,
+      hiveConfig_,
+      ioStats_);
 }
 
 void HiveDataSource::addSplit(std::shared_ptr<ConnectorSplit> split) {