Allow rename and deletion of subfields

Author: rui-mo

velox/connectors/hive/HiveConfig.cpp

@@ -110,6 +110,20 @@ bool HiveConfig::isParquetUseColumnNames(
       config_->get<bool>(kParquetUseColumnNames, false));
 }
 
+bool HiveConfig::isOrcAllowEnhancedSchemaEvolution(
+    const config::ConfigBase* session) const {
+  return session->get<bool>(
+      kOrcAllowEnhancedSchemaEvolutionSession,
+      config_->get<bool>(kOrcAllowEnhancedSchemaEvolution, false));
+}
+
+bool HiveConfig::isParquetAllowEnhancedSchemaEvolution(
+    const config::ConfigBase* session) const {
+  return session->get<bool>(
+      kParquetAllowEnhancedSchemaEvolutionSession,
+      config_->get<bool>(kParquetAllowEnhancedSchemaEvolution, false));
+}
+
 bool HiveConfig::isFileColumnNamesReadAsLowerCase(
     const config::ConfigBase* session) const {
   return session->get<bool>(

velox/connectors/hive/HiveConfig.h

@@ -85,6 +85,16 @@ class HiveConfig {
   static constexpr const char* kParquetUseColumnNamesSession =
       "parquet_use_column_names";
 
+  static constexpr const char* kOrcAllowEnhancedSchemaEvolution =
+      "hive.orc.allow-enhanced-schema-evolution";
+  static constexpr const char* kOrcAllowEnhancedSchemaEvolutionSession =
+      "orc_allow_enhanced_schema_evolution";
+
+  static constexpr const char* kParquetAllowEnhancedSchemaEvolution =
+      "hive.parquet.allow-enhanced-schema-evolution";
+  static constexpr const char* kParquetAllowEnhancedSchemaEvolutionSession =
+      "parquet_allow_enhanced_schema_evolution";
+
   /// Reads the source file column name as lower case.
   static constexpr const char* kFileColumnNamesReadAsLowerCase =
       "file-column-names-read-as-lower-case";
@@ -231,6 +241,12 @@ class HiveConfig {
 
   bool isParquetUseColumnNames(const config::ConfigBase* session) const;
 
+  bool isOrcAllowEnhancedSchemaEvolution(
+      const config::ConfigBase* session) const;
+
+  bool isParquetAllowEnhancedSchemaEvolution(
+      const config::ConfigBase* session) const;
+
   bool isFileColumnNamesReadAsLowerCase(
       const config::ConfigBase* session) const;
 

velox/connectors/hive/HiveConnectorUtil.cpp

@@ -556,24 +556,30 @@ void configureReaderOptions(
       hiveConfig->isFileColumnNamesReadAsLowerCase(sessionProperties));
   readerOptions.setAllowEmptyFile(true);
   bool useColumnNamesForColumnMapping = false;
+  bool allowEnhancedSchemaEvolution = false;
   switch (hiveSplit->fileFormat) {
     case dwio::common::FileFormat::DWRF:
     case dwio::common::FileFormat::ORC: {
       useColumnNamesForColumnMapping =
           hiveConfig->isOrcUseColumnNames(sessionProperties);
+      allowEnhancedSchemaEvolution =
+          hiveConfig->isOrcAllowEnhancedSchemaEvolution(sessionProperties);
       break;
     }
     case dwio::common::FileFormat::PARQUET: {
       useColumnNamesForColumnMapping =
           hiveConfig->isParquetUseColumnNames(sessionProperties);
+      allowEnhancedSchemaEvolution =
+          hiveConfig->isParquetAllowEnhancedSchemaEvolution(sessionProperties);
       break;
     }
     default:
-      useColumnNamesForColumnMapping = false;
+      break;
   }
 
   readerOptions.setUseColumnNamesForColumnMapping(
       useColumnNamesForColumnMapping);
+  readerOptions.setAllowEnhancedSchemaEvolution(allowEnhancedSchemaEvolution);
   readerOptions.setFileSchema(fileSchema);
   readerOptions.setFooterEstimatedSize(hiveConfig->footerEstimatedSize());
   readerOptions.setFilePreloadThreshold(hiveConfig->filePreloadThreshold());

velox/connectors/hive/SplitReader.cpp

@@ -440,12 +440,22 @@ std::vector<TypePtr> SplitReader::adaptColumns(
       auto fileTypeIdx = fileType->getChildIdxIfExists(fieldName);
       if (!fileTypeIdx.has_value()) {
         // Column is missing. Most likely due to schema evolution.
-        VELOX_CHECK(tableSchema, "Unable to resolve column '{}'", fieldName);
+        TypePtr fieldType = nullptr;
+        if (baseReaderOpts_.allowEnhancedSchemaEvolution()) {
+          auto outputTypeIdx =
+              readerOutputType_->getChildIdxIfExists(fieldName);
+          if (outputTypeIdx.has_value()) {
+            // Field name exists in the user-specified output type.
+            fieldType = readerOutputType_->childAt(outputTypeIdx.value());
+          }
+        }
+        if (fieldType == nullptr) {
+          VELOX_CHECK(tableSchema, "Unable to resolve column '{}'", fieldName);
+          fieldType = tableSchema->findChild(fieldName);
+        }
         childSpec->setConstantValue(
             BaseVector::createNullConstant(
-                tableSchema->findChild(fieldName),
-                1,
-                connectorQueryCtx_->memoryPool()));
+                fieldType, 1, connectorQueryCtx_->memoryPool()));
       } else {
         // Column no longer missing, reset constant value set on the spec.
         childSpec->setConstantValue(nullptr);

velox/dwio/common/Options.cpp

@@ -74,6 +74,8 @@ ColumnReaderOptions makeColumnReaderOptions(const ReaderOptions& options) {
   ColumnReaderOptions columnReaderOptions;
   columnReaderOptions.useColumnNamesForColumnMapping_ =
       options.useColumnNamesForColumnMapping();
+  columnReaderOptions.allowEnhancedSchemaEvolution_ =
+      options.allowEnhancedSchemaEvolution();
   return columnReaderOptions;
 }
 

velox/dwio/common/Options.h

@@ -600,6 +600,11 @@ class ReaderOptions : public io::ReaderOptions {
     return *this;
   }
 
+  ReaderOptions& setAllowEnhancedSchemaEvolution(bool flag) {
+    allowEnhancedSchemaEvolution_ = flag;
+    return *this;
+  }
+
   ReaderOptions& setIOExecutor(std::shared_ptr<folly::Executor> executor) {
     ioExecutor_ = std::move(executor);
     return *this;
@@ -675,6 +680,10 @@ class ReaderOptions : public io::ReaderOptions {
     return useColumnNamesForColumnMapping_;
   }
 
+  bool allowEnhancedSchemaEvolution() const {
+    return allowEnhancedSchemaEvolution_;
+  }
+
   const std::shared_ptr<random::RandomSkipTracker>& randomSkip() const {
     return randomSkip_;
   }
@@ -726,6 +735,7 @@ class ReaderOptions : public io::ReaderOptions {
   uint64_t filePreloadThreshold_{kDefaultFilePreloadThreshold};
   bool fileColumnNamesReadAsLowerCase_{false};
   bool useColumnNamesForColumnMapping_{false};
+  bool allowEnhancedSchemaEvolution_{false};
   std::shared_ptr<folly::Executor> ioExecutor_;
   std::shared_ptr<random::RandomSkipTracker> randomSkip_;
   std::shared_ptr<velox::common::ScanSpec> scanSpec_;
@@ -771,6 +781,10 @@ struct ColumnReaderOptions {
   // Whether to map table field names to file field names using names, not
   // indices.
   bool useColumnNamesForColumnMapping_{false};
+
+  // Whether enhanced schema evolution operations, such as renaming or deleting
+  // fields and creating NULL values, are allowed.
+  bool allowEnhancedSchemaEvolution_{false};
 };
 
 ColumnReaderOptions makeColumnReaderOptions(const ReaderOptions& options);

velox/dwio/common/SelectiveFlatMapColumnReader.h

@@ -24,11 +24,13 @@ namespace facebook::velox::dwio::common {
 class SelectiveFlatMapColumnReader : public SelectiveStructColumnReaderBase {
  protected:
   SelectiveFlatMapColumnReader(
+      const dwio::common::ColumnReaderOptions& columnReaderOptions,
       const TypePtr& requestedType,
       const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
       FormatParams& params,
       velox::common::ScanSpec& scanSpec)
       : SelectiveStructColumnReaderBase(
+            columnReaderOptions,
             requestedType,
             fileType,
             params,

velox/dwio/common/SelectiveStructColumnReader.cpp

@@ -425,7 +425,9 @@ void SelectiveStructColumnReaderBase::read(
   }
 
   const auto& childSpecs = scanSpec_->children();
-  VELOX_CHECK(!childSpecs.empty());
+  if (!columnReaderOptions_.allowEnhancedSchemaEvolution_) {
+    VELOX_CHECK(!childSpecs.empty());
+  }
   for (size_t i = 0; i < childSpecs.size(); ++i) {
     const auto& childSpec = childSpecs[i];
 
@@ -512,6 +514,9 @@ void SelectiveStructColumnReaderBase::recordParentNullsInChildren(
 
 bool SelectiveStructColumnReaderBase::isChildMissing(
     const velox::common::ScanSpec& childSpec) const {
+  bool enhancedSchemaEvolutionByName =
+      columnReaderOptions_.useColumnNamesForColumnMapping_ &&
+      columnReaderOptions_.allowEnhancedSchemaEvolution_;
   return
       // The below check is trying to determine if this is a missing field in a
       // struct that should be constant null.
@@ -524,9 +529,12 @@ bool SelectiveStructColumnReaderBase::isChildMissing(
                                                   // row type that doesn't exist
                                                   // in the output.
        fileType_->type()->kind() !=
-           TypeKind::MAP && // If this is the case it means this is a flat map,
-                            // so it can't have "missing" fields.
-       childSpec.channel() >= fileType_->size());
+           TypeKind::MAP // If this is the case it means this is a flat map,
+                         // so it can't have "missing" fields.
+       ) &&
+      (enhancedSchemaEvolutionByName
+           ? !asRowType(fileType_->type())->containsChild(childSpec.fieldName())
+           : childSpec.channel() >= fileType_->size());
 }
 
 std::unique_ptr<velox::dwio::common::ColumnLoader>
@@ -538,7 +546,9 @@ SelectiveStructColumnReaderBase::makeColumnLoader(vector_size_t index) {
 void SelectiveStructColumnReaderBase::getValues(
     const RowSet& rows,
     VectorPtr* result) {
-  VELOX_CHECK(!scanSpec_->children().empty());
+  if (!columnReaderOptions_.allowEnhancedSchemaEvolution_) {
+    VELOX_CHECK(!scanSpec_->children().empty());
+  }
   VELOX_CHECK_NOT_NULL(
       *result, "SelectiveStructColumnReaderBase expects a non-null result");
   VELOX_CHECK(

velox/dwio/common/SelectiveStructColumnReader.h

@@ -16,6 +16,7 @@
 
 #pragma once
 
+#include "velox/dwio/common/Options.h"
 #include "velox/dwio/common/SelectiveColumnReaderInternal.h"
 
 namespace facebook::velox::dwio::common {
@@ -113,13 +114,15 @@ class SelectiveStructColumnReaderBase : public SelectiveColumnReader {
   static constexpr int32_t kConstantChildSpecSubscript{-1};
 
   SelectiveStructColumnReaderBase(
+      const dwio::common::ColumnReaderOptions& columnReaderOptions,
       const TypePtr& requestedType,
       const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
       FormatParams& params,
       velox::common::ScanSpec& scanSpec,
       bool isRoot = false,
       bool generateLazyChildren = true)
       : SelectiveColumnReader(requestedType, fileType, params, scanSpec),
+        columnReaderOptions_(columnReaderOptions),
         debugString_(
             getExceptionContext().message(VeloxException::Type::kSystem)),
         isRoot_(isRoot),
@@ -180,6 +183,8 @@ class SelectiveStructColumnReaderBase : public SelectiveColumnReader {
     }
   }
 
+  const dwio::common::ColumnReaderOptions& columnReaderOptions_;
+
   // Context information obtained from ExceptionContext. Stored here
   // so that LazyVector readers under this can add this to their
   // ExceptionContext. Allows contextualizing reader errors to split

velox/dwio/dwrf/reader/DwrfReader.cpp

@@ -320,13 +320,14 @@ DwrfRowReader::DwrfRowReader(
     makeProjectedNodes(*getReader().schemaWithId(), *projectedNodes_);
   }
 
+  // Reader options must be configured before calling 'getUnitLoader()',
+  // which triggers 'SelectiveDwrfReader::build'.
+  columnReaderOptions_ = dwio::common::makeColumnReaderOptions(
+      readerBaseShared()->readerOptions());
   unitLoader_ = getUnitLoader();
   if (!emptyFile()) {
     getReader().loadCache();
   }
-
-  columnReaderOptions_ = dwio::common::makeColumnReaderOptions(
-      readerBaseShared()->readerOptions());
 }
 
 std::unique_ptr<ColumnReader>& DwrfRowReader::getColumnReader() {