Fix read iceberg table decimal column error

Author: PingLiuPing

velox/connectors/hive/SplitReader.h

@@ -165,18 +165,18 @@ class SplitReader {
       VectorPtr& output,
       const std::vector<BaseVector::CopyRange>& ranges);
 
+  void setPartitionValue(
+      common::ScanSpec* spec,
+      const std::string& partitionKey,
+      const std::optional<std::string>& value) const;
+
  private:
   /// Different table formats may have different meatadata columns.
   /// This function will be used to update the scanSpec for these columns.
-  std::vector<TypePtr> adaptColumns(
+  virtual std::vector<TypePtr> adaptColumns(
       const RowTypePtr& fileType,
       const std::shared_ptr<const velox::RowType>& tableSchema) const;
 
-  void setPartitionValue(
-      common::ScanSpec* spec,
-      const std::string& partitionKey,
-      const std::optional<std::string>& value) const;
-
  protected:
   std::shared_ptr<const HiveConnectorSplit> hiveSplit_;
   const std::shared_ptr<const HiveTableHandle> hiveTableHandle_;

velox/connectors/hive/iceberg/IcebergSplitReader.cpp

@@ -240,4 +240,44 @@ uint64_t IcebergSplitReader::next(uint64_t size, VectorPtr& output) {
   return rowsScanned;
 }
 
+std::vector<TypePtr> IcebergSplitReader::adaptColumns(
+    const RowTypePtr& fileType,
+    const std::shared_ptr<const velox::RowType>& tableSchema) const {
+  // Keep track of schema types for columns in file, used by ColumnSelector.
+  std::vector<TypePtr> columnTypes = fileType->children();
+  auto& childrenSpecs = scanSpec_->children();
+  // Iceberg table stores all column's data in data file.
+  for (size_t i = 0; i < childrenSpecs.size(); ++i) {
+    auto* childSpec = childrenSpecs[i].get();
+    const std::string& fieldName = childSpec->fieldName();
+    auto fileTypeIdx = fileType->getChildIdxIfExists(fieldName);
+    auto outputTypeIdx = readerOutputType_->getChildIdxIfExists(fieldName);
+    if (outputTypeIdx.has_value() && fileTypeIdx.has_value()) {
+      childSpec->setConstantValue(nullptr);
+      auto& outputType = readerOutputType_->childAt(*outputTypeIdx);
+      columnTypes[*fileTypeIdx] = outputType;
+    } else if (!fileTypeIdx.has_value()) {
+      // Handle columns missing from the data file in two scenarios:
+      // 1. Schema evolution: Column was added after the data file was written
+      //    and doesn't exist in older data files.
+      // 2. Partition columns: In Hive-written data files, partition column
+      //    values are stored in partition metadata rather than in the data
+      //    file itself, following Hive's partitioning convention.
+      if (auto it = hiveSplit_->partitionKeys.find(fieldName);
+          it != hiveSplit_->partitionKeys.end()) {
+        setPartitionValue(childSpec, fieldName, it->second);
+      } else {
+        childSpec->setConstantValue(BaseVector::createNullConstant(
+            tableSchema->findChild(fieldName),
+            1,
+            connectorQueryCtx_->memoryPool()));
+      }
+    }
+  }
+
+  scanSpec_->resetCachedValues(false);
+
+  return columnTypes;
+}
+
 } // namespace facebook::velox::connector::hive::iceberg

velox/connectors/hive/iceberg/IcebergSplitReader.h

@@ -55,6 +55,10 @@ class IcebergSplitReader : public SplitReader {
   std::shared_ptr<const dwio::common::TypeWithId> baseFileSchema();
 
  private:
+  std::vector<TypePtr> adaptColumns(
+      const RowTypePtr& fileType,
+      const std::shared_ptr<const velox::RowType>& tableSchema) const override;
+
   // The read offset to the beginning of the split in number of rows for the
   // current batch for the base data file
   uint64_t baseReadOffset_;