Passing 'skip.header.line.count' to SerDeOptions::nullString. (facebo…

…okincubator#8722)

Summary:
Pull Request resolved: facebookincubator#8722

Some TEXT tables have custom NULL string.
We need to pass it to the SerDeOptions::nullString to return
the correct results.

Reviewed By: amitkdutta

Differential Revision: D53643548

fbshipit-source-id: 920f7f3b719e033fe8c31bcdd8dfe00fac405015

velox/connectors/hive/HiveConnectorUtil.cpp

@@ -377,7 +377,8 @@ std::shared_ptr<common::ScanSpec> makeScanSpec(
 }
 
 std::unique_ptr<dwio::common::SerDeOptions> parseSerdeParameters(
-    const std::unordered_map<std::string, std::string>& serdeParameters) {
+    const std::unordered_map<std::string, std::string>& serdeParameters,
+    const std::unordered_map<std::string, std::string>& tableParameters) {
   auto fieldIt = serdeParameters.find(dwio::common::SerDeOptions::kFieldDelim);
   if (fieldIt == serdeParameters.end()) {
     fieldIt = serdeParameters.find("serialization.format");
@@ -393,9 +394,13 @@ std::unique_ptr<dwio::common::SerDeOptions> parseSerdeParameters(
   auto mapKeyIt =
       serdeParameters.find(dwio::common::SerDeOptions::kMapKeyDelim);
 
+  auto nullStringIt = tableParameters.find(
+      dwio::common::TableParameter::kSerializationNullFormat);
+
   if (fieldIt == serdeParameters.end() &&
       collectionIt == serdeParameters.end() &&
-      mapKeyIt == serdeParameters.end()) {
+      mapKeyIt == serdeParameters.end() &&
+      nullStringIt == tableParameters.end()) {
     return nullptr;
   }
 
@@ -413,22 +418,23 @@ std::unique_ptr<dwio::common::SerDeOptions> parseSerdeParameters(
   }
   auto serDeOptions = std::make_unique<dwio::common::SerDeOptions>(
       fieldDelim, collectionDelim, mapKeyDelim);
+  serDeOptions->nullString = nullStringIt->second;
   return serDeOptions;
 }
 
 void configureReaderOptions(
     dwio::common::ReaderOptions& readerOptions,
     const std::shared_ptr<HiveConfig>& hiveConfig,
     const Config* sessionProperties,
-    const RowTypePtr& fileSchema,
-    std::shared_ptr<HiveConnectorSplit> hiveSplit) {
+    const std::shared_ptr<HiveTableHandle>& hiveTableHandle,
+    const std::shared_ptr<HiveConnectorSplit>& hiveSplit) {
   readerOptions.setMaxCoalesceBytes(hiveConfig->maxCoalescedBytes());
   readerOptions.setMaxCoalesceDistance(hiveConfig->maxCoalescedDistanceBytes());
   readerOptions.setFileColumnNamesReadAsLowerCase(
       hiveConfig->isFileColumnNamesReadAsLowerCase(sessionProperties));
   readerOptions.setUseColumnNamesForColumnMapping(
       hiveConfig->isOrcUseColumnNames(sessionProperties));
-  readerOptions.setFileSchema(fileSchema);
+  readerOptions.setFileSchema(hiveTableHandle->dataColumns());
   readerOptions.setFooterEstimatedSize(hiveConfig->footerEstimatedSize());
   readerOptions.setFilePreloadThreshold(hiveConfig->filePreloadThreshold());
 
@@ -439,7 +445,8 @@ void configureReaderOptions(
         dwio::common::toString(readerOptions.getFileFormat()),
         dwio::common::toString(hiveSplit->fileFormat));
   } else {
-    auto serDeOptions = parseSerdeParameters(hiveSplit->serdeParameters);
+    auto serDeOptions = parseSerdeParameters(
+        hiveSplit->serdeParameters, hiveTableHandle->tableParameters());
     if (serDeOptions) {
       readerOptions.setSerDeOptions(*serDeOptions);
     }

velox/connectors/hive/HiveConnectorUtil.h

@@ -26,6 +26,7 @@
 namespace facebook::velox::connector::hive {
 
 class HiveColumnHandle;
+class HiveTableHandle;
 class HiveConfig;
 struct HiveConnectorSplit;
 
@@ -57,8 +58,8 @@ void configureReaderOptions(
     dwio::common::ReaderOptions& readerOptions,
     const std::shared_ptr<HiveConfig>& config,
     const Config* sessionProperties,
-    const RowTypePtr& fileSchema,
-    std::shared_ptr<HiveConnectorSplit> hiveSplit);
+    const std::shared_ptr<HiveTableHandle>& hiveTableHandle,
+    const std::shared_ptr<HiveConnectorSplit>& hiveSplit);
 
 void configureRowReaderOptions(
     dwio::common::RowReaderOptions& rowReaderOptions,

velox/connectors/hive/SplitReader.cpp

@@ -82,7 +82,7 @@ void SplitReader::configureReaderOptions() {
       baseReaderOpts_,
       hiveConfig_,
       connectorQueryCtx_->sessionProperties(),
-      hiveTableHandle_->dataColumns(),
+      hiveTableHandle_,
       hiveSplit_);
 }
 

velox/dwio/common/Options.h

@@ -94,6 +94,8 @@ class SerDeOptions {
 
 struct TableParameter {
   static constexpr const char* kSkipHeaderLineCount = "skip.header.line.count";
+  static constexpr const char* kSerializationNullFormat =
+      "serialization.null.format";
 };
 
 /**