Fix ORC related failed UT (#417)

Author: chenxu14

velox/connectors/hive/HiveConnector.cpp

@@ -140,6 +140,7 @@ std::unique_ptr<core::PartitionFunction> HivePartitionFunctionSpec::create(
 void HiveConnectorFactory::initialize() {
   static bool once = []() {
     dwio::common::registerFileSinks();
+    dwrf::registerOrcReaderFactory();
     dwrf::registerDwrfReaderFactory();
     dwrf::registerDwrfWriterFactory();
 // Meta's buck build system needs this check.

velox/dwio/dwrf/common/FileMetadata.h

@@ -426,7 +426,8 @@ class FooterWrapper : public ProtoWrapperBase {
 
   // TODO: ORC has not supported column statistics yet
   int statisticsSize() const {
-    return format_ == DwrfFormat::kDwrf ? dwrfPtr()->statistics_size() : 0;
+    return format_ == DwrfFormat::kDwrf ? dwrfPtr()->statistics_size()
+                                        : orcPtr()->statistics_size();
   }
 
   const ::google::protobuf::RepeatedPtrField<
@@ -438,7 +439,6 @@ class FooterWrapper : public ProtoWrapperBase {
 
   const ::facebook::velox::dwrf::proto::ColumnStatistics& statistics(
       int index) const {
-    VELOX_CHECK_EQ(format_, DwrfFormat::kDwrf);
     return dwrfPtr()->statistics(index);
   }
 

velox/dwio/dwrf/reader/DwrfReader.cpp

@@ -1108,4 +1108,12 @@ void unregisterDwrfReaderFactory() {
   dwio::common::unregisterReaderFactory(dwio::common::FileFormat::DWRF);
 }
 
+void registerOrcReaderFactory() {
+  dwio::common::registerReaderFactory(std::make_shared<OrcReaderFactory>());
+}
+
+void unregisterOrcReaderFactory() {
+  dwio::common::unregisterReaderFactory(dwio::common::FileFormat::ORC);
+}
+
 } // namespace facebook::velox::dwrf

velox/dwio/dwrf/reader/DwrfReader.h

@@ -365,8 +365,23 @@ class DwrfReaderFactory : public dwio::common::ReaderFactory {
   }
 };
 
+class OrcReaderFactory : public dwio::common::ReaderFactory {
+ public:
+  OrcReaderFactory() : ReaderFactory(dwio::common::FileFormat::ORC) {}
+
+  std::unique_ptr<dwio::common::Reader> createReader(
+      std::unique_ptr<dwio::common::BufferedInput> input,
+      const dwio::common::ReaderOptions& options) override {
+    return DwrfReader::create(std::move(input), options);
+  }
+};
+
 void registerDwrfReaderFactory();
 
 void unregisterDwrfReaderFactory();
 
+void registerOrcReaderFactory();
+
+void unregisterOrcReaderFactory();
+
 } // namespace facebook::velox::dwrf

velox/dwio/dwrf/reader/SelectiveDecimalColumnReader.h

@@ -40,6 +40,10 @@ class SelectiveDecimalColumnReader : public SelectiveColumnReader {
 
   void getValues(RowSet rows, VectorPtr* result) override;
 
+  bool hasBulkPath() const override {
+    return false;
+  }
+
  private:
   template <bool kDense>
   void readHelper(RowSet rows);

velox/dwio/dwrf/reader/SelectiveIntegerDirectColumnReader.h

@@ -63,7 +63,11 @@ class SelectiveIntegerDirectColumnReader
   }
 
   bool hasBulkPath() const override {
-    return true;
+    if (format == velox::dwrf::DwrfFormat::kOrc) {
+      return false; // RLEv2 does't support FastPath yet
+    } else {
+      return true;
+    }
   }
 
   void seekToRowGroup(uint32_t index) override {

velox/dwio/dwrf/reader/SelectiveStringDictionaryColumnReader.h

@@ -53,6 +53,14 @@ class SelectiveStringDictionaryColumnReader
 
   uint64_t skip(uint64_t numValues) override;
 
+  bool hasBulkPath() const override {
+    if (version_ == velox::dwrf::RleVersion_1) {
+      return true;
+    } else {
+      return false; // RLEv2 does't support FastPath yet
+    }
+  }
+
   void read(vector_size_t offset, RowSet rows, const uint64_t* incomingNulls)
       override;