Support struct schema evolution matching by name

rui-mo · Lakehouse Engine Bot · commit d379e1ab4fb4 · 2025-07-03T16:20:53.000Z
Alchemy-item: [Support struct schema evolution matching by name](#29 (comment)) commit 1/1 - 8450c1e
diff --git a/velox/connectors/hive/SplitReader.cpp b/velox/connectors/hive/SplitReader.cpp
@@ -413,11 +413,17 @@ 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);
+        auto outputTypeIdx = readerOutputType_->getChildIdxIfExists(fieldName);
+        TypePtr fieldType;
+        if (outputTypeIdx.has_value()) {
+          // Field name exists in the user-specified output type.
+          fieldType = readerOutputType_->childAt(outputTypeIdx.value());
+        } else {
+          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);
diff --git a/velox/dwio/common/ScanSpec.cpp b/velox/dwio/common/ScanSpec.cpp
@@ -142,7 +142,7 @@ bool ScanSpec::hasFilter() const {
   if (hasFilter_.has_value()) {
     return hasFilter_.value();
   }
-  if (!isConstant() && filter()) {
+  if (filter()) {
     hasFilter_ = true;
     return true;
   }
diff --git a/velox/dwio/common/SelectiveStructColumnReader.cpp b/velox/dwio/common/SelectiveStructColumnReader.cpp
@@ -358,7 +358,6 @@ void SelectiveStructColumnReaderBase::read(
   }
 
   const auto& childSpecs = scanSpec_->children();
-  VELOX_CHECK(!childSpecs.empty());
   for (size_t i = 0; i < childSpecs.size(); ++i) {
     const auto& childSpec = childSpecs[i];
     VELOX_TRACE_HISTORY_PUSH("read %s", childSpec->fieldName().c_str());
@@ -458,15 +457,17 @@ 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.
+       ) &&
+      (columnReaderOptions_.useColumnNamesForColumnMapping_
+           ? !asRowType(fileType_->type())->containsChild(childSpec.fieldName())
+           : childSpec.channel() >= fileType_->size());
 }
 
 void SelectiveStructColumnReaderBase::getValues(
     const RowSet& rows,
     VectorPtr* result) {
-  VELOX_CHECK(!scanSpec_->children().empty());
   VELOX_CHECK_NOT_NULL(
       *result, "SelectiveStructColumnReaderBase expects a non-null result");
   VELOX_CHECK(
diff --git a/velox/dwio/common/SelectiveStructColumnReader.h b/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 {
@@ -111,13 +112,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),
@@ -155,6 +158,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
diff --git a/velox/dwio/dwrf/reader/SelectiveFlatMapColumnReader.cpp b/velox/dwio/dwrf/reader/SelectiveFlatMapColumnReader.cpp
@@ -177,6 +177,7 @@ class SelectiveFlatMapAsStructReader : public SelectiveStructColumnReaderBase {
       DwrfParams& params,
       common::ScanSpec& scanSpec)
       : SelectiveStructColumnReaderBase(
+            columnReaderOptions,
             requestedType,
             fileType,
             params,
@@ -215,6 +216,7 @@ class SelectiveFlatMapReader : public SelectiveStructColumnReaderBase {
       DwrfParams& params,
       common::ScanSpec& scanSpec)
       : SelectiveStructColumnReaderBase(
+            columnReaderOptions,
             requestedType,
             fileType,
             params,
diff --git a/velox/dwio/dwrf/reader/SelectiveStructColumnReader.cpp b/velox/dwio/dwrf/reader/SelectiveStructColumnReader.cpp
@@ -31,6 +31,7 @@ SelectiveStructColumnReader::SelectiveStructColumnReader(
     common::ScanSpec& scanSpec,
     bool isRoot)
     : SelectiveStructColumnReaderBase(
+          columnReaderOptions,
           requestedType,
           fileType,
           params,
diff --git a/velox/dwio/dwrf/reader/SelectiveStructColumnReader.h b/velox/dwio/dwrf/reader/SelectiveStructColumnReader.h
@@ -25,12 +25,14 @@ class SelectiveStructColumnReaderBase
     : public dwio::common::SelectiveStructColumnReaderBase {
  public:
   SelectiveStructColumnReaderBase(
+      const dwio::common::ColumnReaderOptions& columnReaderOptions,
       const TypePtr& requestedType,
       const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
       DwrfParams& params,
       common::ScanSpec& scanSpec,
       bool isRoot = false)
       : dwio::common::SelectiveStructColumnReaderBase(
+            columnReaderOptions,
             requestedType,
             fileType,
             params,
diff --git a/velox/dwio/parquet/reader/ParquetColumnReader.cpp b/velox/dwio/parquet/reader/ParquetColumnReader.cpp
@@ -38,7 +38,8 @@ std::unique_ptr<dwio::common::SelectiveColumnReader> ParquetColumnReader::build(
     const TypePtr& requestedType,
     const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
     ParquetParams& params,
-    common::ScanSpec& scanSpec) {
+    common::ScanSpec& scanSpec,
+    memory::MemoryPool& pool) {
   auto colName = scanSpec.fieldName();
 
   switch (fileType->type()->kind()) {
@@ -59,19 +60,19 @@ std::unique_ptr<dwio::common::SelectiveColumnReader> ParquetColumnReader::build(
 
     case TypeKind::ROW:
       return std::make_unique<StructColumnReader>(
-          columnReaderOptions, requestedType, fileType, params, scanSpec);
+          columnReaderOptions, requestedType, fileType, params, scanSpec, pool);
 
     case TypeKind::VARBINARY:
     case TypeKind::VARCHAR:
       return std::make_unique<StringColumnReader>(fileType, params, scanSpec);
 
     case TypeKind::ARRAY:
       return std::make_unique<ListColumnReader>(
-          columnReaderOptions, requestedType, fileType, params, scanSpec);
+          columnReaderOptions, requestedType, fileType, params, scanSpec, pool);
 
     case TypeKind::MAP:
       return std::make_unique<MapColumnReader>(
-          columnReaderOptions, requestedType, fileType, params, scanSpec);
+          columnReaderOptions, requestedType, fileType, params, scanSpec, pool);
 
     case TypeKind::BOOLEAN:
       return std::make_unique<BooleanColumnReader>(
diff --git a/velox/dwio/parquet/reader/ParquetColumnReader.h b/velox/dwio/parquet/reader/ParquetColumnReader.h
@@ -47,6 +47,7 @@ class ParquetColumnReader {
       const TypePtr& requestedType,
       const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
       ParquetParams& params,
-      common::ScanSpec& scanSpec);
+      common::ScanSpec& scanSpec,
+      memory::MemoryPool& pool);
 };
 } // namespace facebook::velox::parquet
diff --git a/velox/dwio/parquet/reader/ParquetReader.cpp b/velox/dwio/parquet/reader/ParquetReader.cpp
@@ -1066,12 +1066,15 @@ class ParquetRowReader::Impl {
         options_.timestampPrecision());
     requestedType_ = options_.requestedType() ? options_.requestedType()
                                               : readerBase_->schema();
+    columnReaderOptions_ =
+        dwio::common::makeColumnReaderOptions(readerBase_->options());
     columnReader_ = ParquetColumnReader::build(
         columnReaderOptions_,
         requestedType_,
         readerBase_->schemaWithId(), // Id is schema id
         params,
-        *options_.scanSpec());
+        *options_.scanSpec(),
+        pool_);
     columnReader_->setIsTopLevel();
 
     filterRowGroups();
@@ -1081,9 +1084,6 @@ class ParquetRowReader::Impl {
       // table scan.
       advanceToNextRowGroup();
     }
-
-    columnReaderOptions_ =
-        dwio::common::makeColumnReaderOptions(readerBase_->options());
   }
 
   void filterRowGroups() {
diff --git a/velox/dwio/parquet/reader/RepeatedColumnReader.cpp b/velox/dwio/parquet/reader/RepeatedColumnReader.cpp
@@ -33,6 +33,9 @@ PageReader* readLeafRepDefs(
       return nullptr;
     }
     auto pageReader = reader->formatData().as<ParquetData>().reader();
+    if (pageReader == nullptr) {
+      return nullptr;
+    }
     pageReader->decodeRepDefs(numTop);
     return pageReader;
   }
@@ -114,7 +117,8 @@ MapColumnReader::MapColumnReader(
     const TypePtr& requestedType,
     const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
     ParquetParams& params,
-    common::ScanSpec& scanSpec)
+    common::ScanSpec& scanSpec,
+    memory::MemoryPool& pool)
     : dwio::common::SelectiveMapColumnReader(
           requestedType,
           fileType,
@@ -128,13 +132,15 @@ MapColumnReader::MapColumnReader(
       keyChildType,
       fileType_->childAt(0),
       params,
-      *scanSpec.children()[0]);
+      *scanSpec.children()[0],
+      pool);
   elementReader_ = ParquetColumnReader::build(
       columnReaderOptions,
       elementChildType,
       fileType_->childAt(1),
       params,
-      *scanSpec.children()[1]);
+      *scanSpec.children()[1],
+      pool);
   reinterpret_cast<const ParquetTypeWithId*>(fileType.get())
       ->makeLevelInfo(levelInfo_);
   children_ = {keyReader_.get(), elementReader_.get()};
@@ -232,7 +238,8 @@ ListColumnReader::ListColumnReader(
     const TypePtr& requestedType,
     const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
     ParquetParams& params,
-    common::ScanSpec& scanSpec)
+    common::ScanSpec& scanSpec,
+    memory::MemoryPool& pool)
     : dwio::common::SelectiveListColumnReader(
           requestedType,
           fileType,
@@ -244,7 +251,8 @@ ListColumnReader::ListColumnReader(
       childType,
       fileType_->childAt(0),
       params,
-      *scanSpec.children()[0]);
+      *scanSpec.children()[0],
+      pool);
   reinterpret_cast<const ParquetTypeWithId*>(fileType.get())
       ->makeLevelInfo(levelInfo_);
   children_ = {child_.get()};
diff --git a/velox/dwio/parquet/reader/RepeatedColumnReader.h b/velox/dwio/parquet/reader/RepeatedColumnReader.h
@@ -61,7 +61,8 @@ class MapColumnReader : public dwio::common::SelectiveMapColumnReader {
       const TypePtr& requestedType,
       const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
       ParquetParams& params,
-      common::ScanSpec& scanSpec);
+      common::ScanSpec& scanSpec,
+      memory::MemoryPool& pool);
 
   void prepareRead(
       vector_size_t offset,
@@ -118,7 +119,8 @@ class ListColumnReader : public dwio::common::SelectiveListColumnReader {
       const TypePtr& requestedType,
       const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
       ParquetParams& params,
-      common::ScanSpec& scanSpec);
+      common::ScanSpec& scanSpec,
+      memory::MemoryPool& pool);
 
   void prepareRead(
       vector_size_t offset,
diff --git a/velox/dwio/parquet/reader/StructColumnReader.cpp b/velox/dwio/parquet/reader/StructColumnReader.cpp
@@ -31,18 +31,33 @@ StructColumnReader::StructColumnReader(
     const TypePtr& requestedType,
     const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
     ParquetParams& params,
-    common::ScanSpec& scanSpec)
-    : SelectiveStructColumnReader(requestedType, fileType, params, scanSpec) {
+    common::ScanSpec& scanSpec,
+    memory::MemoryPool& pool)
+    : SelectiveStructColumnReader(
+          columnReaderOptions,
+          requestedType,
+          fileType,
+          params,
+          scanSpec,
+          /*isRoot=*/false) {
   auto& childSpecs = scanSpec_->stableChildren();
+  const bool useColumnNames =
+      columnReaderOptions.useColumnNamesForColumnMapping_;
+  std::vector<column_index_t> missingFields;
   for (auto i = 0; i < childSpecs.size(); ++i) {
     auto childSpec = childSpecs[i];
-    if (childSpec->isConstant() || isChildMissing(*childSpec)) {
+    if (childSpec->isConstant() &&
+        (!useColumnNames && isChildMissing(*childSpec))) {
       childSpec->setSubscript(kConstantChildSpecSubscript);
       continue;
     }
     if (!childSpecs[i]->readFromFile()) {
       continue;
     }
+    if (useColumnNames && isChildMissing(*childSpec)) {
+      missingFields.emplace_back(i);
+      continue;
+    }
     auto childFileType = fileType_->childByName(childSpec->fieldName());
     auto childRequestedType =
         requestedType_->asRow().findChild(childSpec->fieldName());
@@ -51,10 +66,30 @@ StructColumnReader::StructColumnReader(
         childRequestedType,
         childFileType,
         params,
-        *childSpec));
+        *childSpec,
+        pool));
 
     childSpecs[i]->setSubscript(children_.size() - 1);
   }
+
+  // 'missingFields' is not empty only when using column names for column
+  // mapping.
+  if (missingFields.size() > 0) {
+    // Set the struct as null if all the subfields in the requested type are
+    // missing and the number of subfields is more than one.
+    if (childSpecs.size() > 1 && missingFields.size() == childSpecs.size()) {
+      scanSpec_->setConstantValue(
+          BaseVector::createNullConstant(requestedType_, 1, &pool));
+    } else {
+      // Set null constant for the missing subfield of requested type.
+      auto rowTypePtr = asRowType(requestedType_);
+      for (int channel : missingFields) {
+        childSpecs[channel]->setConstantValue(BaseVector::createNullConstant(
+            rowTypePtr->findChild(childSpecs[channel]->fieldName()), 1, &pool));
+      }
+    }
+  }
+
   auto type = reinterpret_cast<const ParquetTypeWithId*>(fileType_.get());
   if (type->parent()) {
     levelMode_ = reinterpret_cast<const ParquetTypeWithId*>(fileType_.get())
@@ -64,7 +99,10 @@ StructColumnReader::StructColumnReader(
     // this and the child.
     auto child = childForRepDefs_;
     for (;;) {
-      assert(child);
+      if (child == nullptr) {
+        levelMode_ = LevelMode::kNulls;
+        break;
+      }
       if (child->fileType().type()->kind() == TypeKind::ARRAY ||
           child->fileType().type()->kind() == TypeKind::MAP) {
         levelMode_ = LevelMode::kStructOverLists;
@@ -101,7 +139,6 @@ StructColumnReader::findBestLeaf() {
       best = child;
     }
   }
-  assert(best);
   return best;
 }
 
diff --git a/velox/dwio/parquet/reader/StructColumnReader.h b/velox/dwio/parquet/reader/StructColumnReader.h
@@ -37,7 +37,8 @@ class StructColumnReader : public dwio::common::SelectiveStructColumnReader {
       const TypePtr& requestedType,
       const std::shared_ptr<const dwio::common::TypeWithId>& fileType,
       ParquetParams& params,
-      common::ScanSpec& scanSpec);
+      common::ScanSpec& scanSpec,
+      memory::MemoryPool& pool);
 
   void read(int64_t offset, const RowSet& rows, const uint64_t* incomingNulls)
       override;
diff --git a/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp b/velox/dwio/parquet/tests/reader/ParquetTableScanTest.cpp
diff --git a/velox/exec/tests/TableScanTest.cpp b/velox/exec/tests/TableScanTest.cpp

Original file line number	Diff line number	Diff line change
`@@ -142,7 +142,7 @@ bool ScanSpec::hasFilter() const {`
`142`	`142`	`if (hasFilter_.has_value()) {`
`143`	`143`	`return hasFilter_.value();`
`144`	`144`	`}`
`145`		`- if (!isConstant() && filter()) {`
	`145`	`+ if (filter()) {`
`146`	`146`	`hasFilter_ = true;`
`147`	`147`	`return true;`
`148`	`148`	`}`