fix: Pass request type to SelectiveDecimalColumnReader

velox/dwio/dwrf/reader/SelectiveDecimalColumnReader.cpp

@@ -20,11 +20,20 @@
 
 template <typename DataT>
 SelectiveDecimalColumnReader<DataT>::SelectiveDecimalColumnReader(
+    const TypePtr& requestedType,
     const std::shared_ptr<const TypeWithId>& fileType,
     DwrfParams& params,
     common::ScanSpec& scanSpec)
-    : SelectiveColumnReader(fileType->type(), fileType, params, scanSpec) {
+    // Read using requestedType so that values are materialized at the
+    // table-schema scale rather than the file-footer scale. See the header
+    // comment for the Hive ORC DECIMAL(38, 18) footer behavior this works
+    // around.
+    : SelectiveColumnReader(requestedType, fileType, params, scanSpec) {
+  VELOX_CHECK(
+      requestedType_->isDecimal(),
+      "SelectiveDecimalColumnReader requires a decimal requestedType, got {}",
+      requestedType_->toString());
   EncodingKey encodingKey{fileType_->id(), params.flatMapContext().sequence};
   auto& stripe = params.stripeStreams();
   if constexpr (std::is_same_v<DataT, std::int64_t>) {
     scale_ = requestedType_->asShortDecimal().scale();

velox/dwio/dwrf/reader/SelectiveDecimalColumnReader.h

@@ -28,7 +28,16 @@ using namespace dwio::common;
 template <typename DataT>
 class SelectiveDecimalColumnReader : public SelectiveColumnReader {
  public:
+  // requestedType is the DECIMAL type to materialize values as. It must be a
+  // decimal type. Hive's ORC writer always records DECIMAL(38, 18) in the
+  // file footer regardless of the metastore-declared precision/scale; the
+  // per-row scale at which each value was actually written lives in the
+  // SECONDARY (a.k.a. NANO_DATA) stream. The reader uses
+  // requestedType.scale() (the table-schema scale) as the target scale and
+  // rescales each value from its per-row scale, so the output matches what
+  // table consumers expect even when the file footer scale differs.
   SelectiveDecimalColumnReader(
+      const TypePtr& requestedType,
       const std::shared_ptr<const TypeWithId>& fileType,
       DwrfParams& params,
       common::ScanSpec& scanSpec);

velox/dwio/dwrf/reader/SelectiveDwrfReader.cpp

@@ -81,7 +81,7 @@ std::unique_ptr<SelectiveColumnReader> SelectiveDwrfReader::build(
     case TypeKind::BIGINT:
       if (fileType->type()->isDecimal()) {
         return std::make_unique<SelectiveDecimalColumnReader<int64_t>>(
-            fileType, params, scanSpec);
+            requestedType, fileType, params, scanSpec);
       } else {
         return buildIntegerReader(
             requestedType, fileType, params, LONG_BYTE_SIZE, scanSpec);
@@ -150,7 +150,7 @@ std::unique_ptr<SelectiveColumnReader> SelectiveDwrfReader::build(
     case TypeKind::HUGEINT:
       if (fileType->type()->isDecimal()) {
         return std::make_unique<SelectiveDecimalColumnReader<int128_t>>(
-            fileType, params, scanSpec);
+            requestedType, fileType, params, scanSpec);
       }
       [[fallthrough]];
     default:

velox/dwio/dwrf/test/TestColumnReader.cpp

@@ -27,6 +27,7 @@
 #include "velox/vector/ComplexVector.h"
 #include "velox/vector/DictionaryVector.h"
 #include "velox/vector/FlatVector.h"
+#include "velox/vector/tests/utils/VectorTestBase.h"
 
 #include <folly/Random.h>
 #include <folly/String.h>
@@ -360,6 +361,53 @@
   bool parallelDecoding() const override {
     return parallelDecoding_;
   }
+
+  // Helper for SelectiveDecimalColumnReader tests that exercise schema
+  // mismatch between the file footer (e.g. Hive ORC's DECIMAL(38, 18))
+  // and the requested table-schema type.
+  template <typename DataT>
+  void verifyDecimalRequestedType(
+      folly::Range<const unsigned char*> dataBuffer,
+      folly::Range<const unsigned char*> scaleBuffer,
+      const RowTypePtr& fileType,
+      const RowTypePtr& requestedType,
+      const std::vector<DataT>& expectedValues) {
+    proto::ColumnEncoding directEncoding;
+    directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+    EXPECT_CALL(streams_, getEncodingProxy(_))
+        .WillRepeatedly(Return(&directEncoding));
+
+    EXPECT_CALL(
+        streams_, getStreamProxy(_, proto::Stream_Kind_ROW_INDEX, false))
+        .WillRepeatedly(Return(nullptr));
+    EXPECT_CALL(streams_, getStreamProxy(_, proto::Stream_Kind_PRESENT, false))
+        .WillRepeatedly(Return(nullptr));
+
+    EXPECT_CALL(streams_, getStreamProxy(1, proto::Stream_Kind_DATA, true))
+        .WillRepeatedly(Return(new SeekableArrayInputStream(
+            dataBuffer.data(), dataBuffer.size())));
+    EXPECT_CALL(streams_, getStreamProxy(1, proto::Stream_Kind_NANO_DATA, true))
+        .WillRepeatedly(Return(new SeekableArrayInputStream(
+            scaleBuffer.data(), scaleBuffer.size())));
+
+    auto scanSpec = std::make_unique<common::ScanSpec>("root");
+    buildReader(requestedType, fileType, {}, scanSpec.get());
+    VectorPtr batch = newBatch(requestedType);
+    skipAndRead(batch, /*read=*/expectedValues.size());
+
+    auto actual = getOnlyChild<FlatVector<DataT>>(batch);
+    ASSERT_EQ(expectedValues.size(), batch->size());
+    ASSERT_EQ(0, getNullCount(batch));
+    ASSERT_EQ(0, getNullCount(actual));
+
+    auto* pool = &streams_.getMemoryPool();
+    auto expected = BaseVector::create<FlatVector<DataT>>(
+        requestedType->childAt(0), expectedValues.size(), pool);
+    for (vector_size_t i = 0; i < expectedValues.size(); ++i) {
+      expected->set(i, expectedValues[i]);
+    }
+    facebook::velox::test::assertEqualVectors(expected, actual);
+  }
 };
 
 struct NonSelectiveReaderTestParams {
@@ -4077,6 +4125,125 @@
       DecimalUtil::toString(intBatch->valueAt(4), decimalType));
 }
 
+// The following tests verify that SelectiveDecimalColumnReader uses
+// requestedType (the metastore/table schema) to determine the target scale,
+// rather than the file-footer scale. See the comment on
+// SelectiveDecimalColumnReader's constructor for the underlying Hive ORC
+// DECIMAL(38, 18) footer behavior these tests cover.
+
+// Integer column (per-row scale=0) stored in an ORC file whose footer
+// declares DECIMAL(38, 18), but the metastore says DECIMAL(20, 0). The
+// reader must produce the original integer values without rescaling.
+TEST_P(TestColumnReader, testDecimal128RequestedTypeScaleZero) {
+  // DATA: values 1, 2, 3, 4 as zigzag-encoded varints.
+  const unsigned char dataBuffer[] = {0x02, 0x04, 0x06, 0x08};
+  // SECONDARY (NANO_DATA): 4 rows with scale=0.
+  // RLEv1 run: ctrl=0x01 (run of 4), delta=0x00, base=zigzag(0)=0x00.
+  const unsigned char scaleBuffer[] = {0x01, 0x00, 0x00};
+  verifyDecimalRequestedType<int128_t>(
+      folly::Range(dataBuffer, std::size(dataBuffer)),
+      folly::Range(scaleBuffer, std::size(scaleBuffer)),
+      ROW({"col_0"}, {DECIMAL(38, 18)}),
+      ROW({"col_0"}, {DECIMAL(20, 0)}),
+      // Expected: original integers, not multiplied by 10^18.
+      {int128_t{1}, int128_t{2}, int128_t{3}, int128_t{4}});
+}
+
+// Per-row scale (5) already matches the requestedType scale (5); no
+// rescaling expected.
+TEST_P(TestColumnReader, testDecimal128RequestedTypeScaleMatchesData) {
+  // DATA: values 1, 2, 3, 4.
+  const unsigned char dataBuffer[] = {0x02, 0x04, 0x06, 0x08};
+  // SECONDARY: 4 rows with scale=5. RLEv1 base=zigzag(5)=0x0A.
+  const unsigned char scaleBuffer[] = {0x01, 0x00, 0x0A};
+  verifyDecimalRequestedType<int128_t>(
+      folly::Range(dataBuffer, std::size(dataBuffer)),
+      folly::Range(scaleBuffer, std::size(scaleBuffer)),
+      ROW({"col_0"}, {DECIMAL(38, 18)}),
+      ROW({"col_0"}, {DECIMAL(25, 5)}),
+      {int128_t{1}, int128_t{2}, int128_t{3}, int128_t{4}});
+}
+
+// Per-row scale (3) is lower than the requestedType scale (5). The reader
+// must upscale by multiplying by 10^(5-3) = 100.
+TEST_P(TestColumnReader, testDecimal128RequestedTypeUpscale) {
+  const unsigned char dataBuffer[] = {0x02, 0x04, 0x06, 0x08};
+  // SECONDARY: scale=3, base=zigzag(3)=0x06.
+  const unsigned char scaleBuffer[] = {0x01, 0x00, 0x06};
+  verifyDecimalRequestedType<int128_t>(
+      folly::Range(dataBuffer, std::size(dataBuffer)),
+      folly::Range(scaleBuffer, std::size(scaleBuffer)),
+      ROW({"col_0"}, {DECIMAL(38, 18)}),
+      ROW({"col_0"}, {DECIMAL(25, 5)}),
+      {int128_t{100}, int128_t{200}, int128_t{300}, int128_t{400}});
+}
+
+// Short decimal (BIGINT, precision<=18). File declares DECIMAL(12, 5),
+// metastore says DECIMAL(10, 2). Reader must downscale by 10^(5-2)=1000.
+TEST_P(TestColumnReader, testDecimal64RequestedTypeDownscale) {
+  // DATA: values 1000, 2000, 3000, 4000 as zigzag varints.
+  const unsigned char dataBuffer[] = {
+      0xD0,
+      0x0F, // 1000
+      0xA0,
+      0x1F, // 2000
+      0xF0,
+      0x2E, // 3000
+      0xC0,
+      0x3E, // 4000
+  };
+  // SECONDARY: scale=5, base=zigzag(5)=0x0A.
+  const unsigned char scaleBuffer[] = {0x01, 0x00, 0x0A};
+  verifyDecimalRequestedType<int64_t>(
+      folly::Range(dataBuffer, std::size(dataBuffer)),
+      folly::Range(scaleBuffer, std::size(scaleBuffer)),
+      ROW({"col_0"}, {DECIMAL(12, 5)}),
+      ROW({"col_0"}, {DECIMAL(10, 2)}),
+      {int64_t{1}, int64_t{2}, int64_t{3}, int64_t{4}});
+}
+
+// A non-decimal requestedType against a decimal file column is unsupported
+// and must be rejected when the reader is constructed.
+//
+// SelectiveDwrfReader::build runs the type-compatibility check first, so the
+// schema-mismatch error is raised before the SelectiveDecimalColumnReader
+// constructor's own VELOX_CHECK runs.
+TEST_P(TestColumnReader, testDecimalRequestedTypeBigintRejected) {
+  proto::ColumnEncoding directEncoding;
+  directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+  EXPECT_CALL(streams_, getEncodingProxy(_))
+      .WillRepeatedly(Return(&directEncoding));
+  EXPECT_CALL(streams_, getStreamProxy(_, proto::Stream_Kind_ROW_INDEX, false))
+      .WillRepeatedly(Return(nullptr));
+  EXPECT_CALL(streams_, getStreamProxy(_, proto::Stream_Kind_PRESENT, false))
+      .WillRepeatedly(Return(nullptr));
+
+  auto fileType = ROW({"col_0"}, {DECIMAL(38, 18)});
+  auto requestedType = ROW({"col_0"}, {BIGINT()});
+  auto scanSpec = std::make_unique<common::ScanSpec>("root");
+  VELOX_ASSERT_THROW(
+      buildReader(requestedType, fileType, {}, scanSpec.get()),
+      "Schema mismatch");
+}
+
+TEST_P(TestColumnReader, testDecimalRequestedTypeDoubleRejected) {
+  proto::ColumnEncoding directEncoding;
+  directEncoding.set_kind(proto::ColumnEncoding_Kind_DIRECT);
+  EXPECT_CALL(streams_, getEncodingProxy(_))
+      .WillRepeatedly(Return(&directEncoding));
+  EXPECT_CALL(streams_, getStreamProxy(_, proto::Stream_Kind_ROW_INDEX, false))
+      .WillRepeatedly(Return(nullptr));
+  EXPECT_CALL(streams_, getStreamProxy(_, proto::Stream_Kind_PRESENT, false))
+      .WillRepeatedly(Return(nullptr));
+
+  auto fileType = ROW({"col_0"}, {DECIMAL(38, 18)});
+  auto requestedType = ROW({"col_0"}, {DOUBLE()});
+  auto scanSpec = std::make_unique<common::ScanSpec>("root");
+  VELOX_ASSERT_THROW(
+      buildReader(requestedType, fileType, {}, scanSpec.get()),
+      "Schema mismatch");
+}
+
 TEST_P(TestColumnReader, testLargeSkip) {
   // set getEncoding
   proto::ColumnEncoding directEncoding;