simoneves · simoneves · Jul 14, 2026 · Jul 14, 2026 · Jul 14, 2026 · Jul 14, 2026
@@ -23,7 +23,6 @@
 #include "velox/experimental/cudf/exec/ToCudf.h"
 #include "velox/experimental/cudf/exec/VeloxCudfInterop.h"
 #include "velox/experimental/cudf/expression/ExpressionEvaluator.h"
-#include "velox/experimental/cudf/expression/SubfieldFiltersToAst.h"
 #include "velox/experimental/cudf/vector/CudfVector.h"
 
 #include "velox/common/time/Timer.h"
@@ -119,13 +118,8 @@ CudfHiveDataSource::CudfHiveDataSource(
     // readColumnNames_
   }
 
-  // Build a combined AST for all subfield filters once. This is query-constant
-  // and doesn't depend on split-specific state.
-  if (!subfieldFilters_.empty()) {
-    auto const readerFilterType = getTableRowType();
-    subfieldFilterExpr_ = &createAstFromSubfieldFilters(
-        subfieldFilters_, subfieldTree_, subfieldScalars_, readerFilterType);
-  }
+  // Subfield filter AST is built per split in CudfSplitReader after reading
+  // the Parquet schema so literal types match libcudf storage types.
 
   VELOX_CHECK_NOT_NULL(fileHandleFactory_, "No FileHandleFactory present");
 
@@ -152,7 +146,20 @@ std::unique_ptr<CudfSplitReader> CudfHiveDataSource::createCudfSplitReader() {
       ioStatistics_,
       ioStats_,
       useExperimentalCudfReader_,
-      subfieldFilterExpr_);
+      makeSubfieldFilterBuildState());
+}
+
+SubfieldFilterBuildState CudfHiveDataSource::makeSubfieldFilterBuildState() {
+  if (subfieldFilters_.empty()) {
+    return {};
+  }
+  return SubfieldFilterBuildState{
+      .filters = &subfieldFilters_,
+      .tree = &subfieldTree_,
+      .scalars = &subfieldScalars_,
+      .rowType = getTableRowType(),
+      .expr = &subfieldFilterExpr_,
+  };
 }
 
 void CudfHiveDataSource::convertSplit(std::shared_ptr<ConnectorSplit> split) {

@@ -111,6 +111,9 @@ class CudfHiveDataSource : public DataSource, public NvtxHelper {
   // Cached combined subfield filter expression owned by 'subfieldTree_'.
   cudf::ast::expression const* subfieldFilterExpr_{nullptr};
 
+  // Build state passed to CudfSplitReader for per-split AST construction.
+  SubfieldFilterBuildState makeSubfieldFilterBuildState();
+
  private:
   // Construct and cache a RowTypePtr for the table column names and types.
   const RowTypePtr getTableRowType();

@@ -18,6 +18,8 @@
 #include "velox/experimental/cudf/connectors/hive/CudfSplitReader.h"
 #include "velox/experimental/cudf/connectors/hive/CudfSplitReaderHelpers.h"
 #include "velox/experimental/cudf/exec/GpuResources.h"
+#include "velox/experimental/cudf/expression/ParquetSchemaUtils.h"
+#include "velox/experimental/cudf/expression/SubfieldFiltersToAst.h"
 
 #include "velox/common/caching/CacheTTLController.h"
 #include "velox/common/time/Timer.h"
@@ -74,7 +76,7 @@ CudfSplitReader::CudfSplitReader(
     const std::shared_ptr<io::IoStatistics>& ioStatistics,
     const std::shared_ptr<IoStats>& ioStats,
     bool useExperimentalCudfReader,
-    cudf::ast::expression const* subfieldFilterExpr)
+    SubfieldFilterBuildState subfieldFilterBuildState)
     : NvtxHelper(
           nvtx3::rgb{80, 171, 241},
           std::nullopt,
@@ -92,7 +94,7 @@ CudfSplitReader::CudfSplitReader(
       pool_(connectorQueryCtx->memoryPool()),
       useExperimentalCudfReader_(useExperimentalCudfReader),
       baseReaderOpts_(pool_),
-      subfieldFilterExpr_(subfieldFilterExpr) {
+      subfieldFilterBuildState_(std::move(subfieldFilterBuildState)) {
   baseReaderOpts_.setDataIoStats(ioStatistics_);
   baseReaderOpts_.setMetadataIoStats(ioStatistics_);
 }
@@ -223,7 +225,44 @@ void CudfSplitReader::resetSplit() {
 }
 
 cudf::ast::expression const* CudfSplitReader::subfieldFilter() {
-  return subfieldFilterExpr_;
+  if (hasSubfieldFilters() && subfieldFilterBuildState_.expr != nullptr &&
+      *subfieldFilterBuildState_.expr == nullptr && dataSource_ != nullptr) {
+    buildSubfieldFilterAst();
+  }
+  if (subfieldFilterBuildState_.expr == nullptr ||
+      *subfieldFilterBuildState_.expr == nullptr) {
+    return nullptr;
+  }
+  return *subfieldFilterBuildState_.expr;
+}
+
+bool CudfSplitReader::hasSubfieldFilters() const {
+  return subfieldFilterBuildState_.filters != nullptr &&
+      !subfieldFilterBuildState_.filters->empty();
+}
+
+void CudfSplitReader::buildSubfieldFilterAst() {
+  auto& state = subfieldFilterBuildState_;
+  if (state.filters == nullptr || state.filters->empty()) {
+    return;
+  }
+  VELOX_CHECK_NOT_NULL(state.tree);
+  VELOX_CHECK_NOT_NULL(state.scalars);
+  VELOX_CHECK_NOT_NULL(state.expr);
+  VELOX_CHECK_NOT_NULL(dataSource_);
+
+  *state.tree = cudf::ast::tree{};
+  state.scalars->clear();
+
+  auto sourceInfo = cudf::io::source_info{dataSource_.get()};
+  auto metadata = cudf::io::read_parquet_metadata(sourceInfo);
+  auto parquetColumnTypes = parquetColumnTypesFromMetadata(metadata);
+  *state.expr = &createAstFromSubfieldFilters(
+      *state.filters,
+      *state.tree,
+      *state.scalars,
+      state.rowType,
+      &parquetColumnTypes);
 }
 
 void CudfSplitReader::setupCudfDataSource() {
@@ -334,6 +373,7 @@ void CudfSplitReader::setupReaderOptions() {
           .allow_mismatched_pq_schemas(
               cudfHiveConfig_->isAllowMismatchedCudfHiveSchemas())
           .timestamp_type(cudfHiveConfig_->timestampType())
+          .case_sensitive_names(false)
           .build();
 
   // Set skip_bytes and num_bytes if available
@@ -344,14 +384,14 @@ void CudfSplitReader::setupReaderOptions() {
     readerOptions_.set_num_bytes(split_->size());
   }
 
-  if (auto* filter = subfieldFilter(); filter != nullptr) {
-    readerOptions_.set_filter(*filter);
-  }
-
-  // Set column projection if needed
   if (readColumnNames_.size()) {
     readerOptions_.set_column_names(readColumnNames_);
   }
+
+  buildSubfieldFilterAst();
+  if (auto* filter = subfieldFilter(); filter != nullptr) {
+    readerOptions_.set_filter(*filter);
+  }
 }
 
 rmm::device_async_resource_ref CudfSplitReader::determineCudfMemoryResource() {

@@ -27,6 +27,7 @@
 #include "velox/connectors/hive/FileHandle.h"
 #include "velox/connectors/hive/TableHandle.h"
 #include "velox/dwio/common/Statistics.h"
+#include "velox/type/Filter.h"
 #include "velox/type/Type.h"
 
 #include <cudf/io/datasource.hpp>
@@ -39,6 +40,16 @@ namespace facebook::velox::cudf_velox::connector::hive {
 
 using namespace facebook::velox::connector;
 
+// Mutable state owned by CudfHiveDataSource; the split reader builds the AST
+// once per split after reading the Parquet schema.
+struct SubfieldFilterBuildState {
+  const common::SubfieldFilters* filters{nullptr};
+  cudf::ast::tree* tree{nullptr};
+  std::vector<std::unique_ptr<cudf::scalar>>* scalars{nullptr};
+  RowTypePtr rowType;
+  cudf::ast::expression const** expr{nullptr};
+};
+
 using CudfParquetReader = cudf::io::chunked_parquet_reader;
 using CudfParquetReaderPtr = std::unique_ptr<CudfParquetReader>;
 
@@ -61,7 +72,7 @@ class CudfSplitReader : public NvtxHelper {
       const std::shared_ptr<io::IoStatistics>& ioStatistics,
       const std::shared_ptr<IoStats>& ioStats,
       bool useExperimentalCudfReader,
-      cudf::ast::expression const* subfieldFilterExpr);
+      SubfieldFilterBuildState subfieldFilterBuildState = {});
 
   virtual ~CudfSplitReader() = default;
 
@@ -84,6 +95,9 @@ class CudfSplitReader : public NvtxHelper {
   // Return the subfield filter.
   virtual cudf::ast::expression const* subfieldFilter();
 
+  // Whether subfield filters were provided for this scan.
+  bool hasSubfieldFilters() const;
+
   // Determine the output memory resource for the cuDF reader.
   virtual rmm::device_async_resource_ref determineCudfMemoryResource();
 
@@ -127,6 +141,9 @@ class CudfSplitReader : public NvtxHelper {
   std::vector<cudf::io::parquet::FileMetaData> fileMetaData_;
 
  private:
+  // Build the subfield filter AST from the current split's Parquet schema.
+  void buildSubfieldFilterAst();
+
   // Setup the cuDF reader options
   void setupReaderOptions();
 
@@ -142,7 +159,7 @@ class CudfSplitReader : public NvtxHelper {
   bool useExperimentalCudfReader_;
 
   dwio::common::ReaderOptions baseReaderOpts_;
-  cudf::ast::expression const* subfieldFilterExpr_;
+  SubfieldFilterBuildState subfieldFilterBuildState_;
 
   struct TotalScanTimeCallbackData {
     uint64_t startTimeUs;

@@ -78,7 +78,7 @@ CudfIcebergDataSource::createCudfSplitReader() {
       ioStatistics_,
       ioStats_,
       useExperimentalCudfReader_,
-      subfieldFilterExpr_);
+      makeSubfieldFilterBuildState());
 }
 
 } // namespace facebook::velox::cudf_velox::connector::hive::iceberg
@@ -93,7 +93,7 @@ CudfIcebergSplitReader::CudfIcebergSplitReader(
     const std::shared_ptr<io::IoStatistics>& ioStatistics,
     const std::shared_ptr<IoStats>& ioStats,
     bool useExperimentalCudfReader,
-    cudf::ast::expression const* subfieldFilterExpr)
+    SubfieldFilterBuildState subfieldFilterBuildState)
     : CudfSplitReader(
           std::move(split),
           std::move(tableHandle),
@@ -106,7 +106,7 @@ CudfIcebergSplitReader::CudfIcebergSplitReader(
           ioStatistics,
           ioStats,
           useExperimentalCudfReader,
-          subfieldFilterExpr),
+          std::move(subfieldFilterBuildState)),
       icebergSplit_(std::move(icebergSplit)),
       hiveConfig_(hiveConfig) {}
 
@@ -144,7 +144,7 @@ void CudfIcebergSplitReader::prepareSplit(
 
   // Defer subfield filter when it cannot evaluate on the physical parquet
   // table, or when positional deletes are present.
-  deferSubfieldFilter_ = CudfSplitReader::subfieldFilter() != nullptr and
+  deferSubfieldFilter_ = hasSubfieldFilters() and
       (noColumnsToRead_ or injectedColumns_.size() or
        // TODO(mh): Drop positional/DV deferral when cudf PR #23077 merges.
        deletionVectorReader_ or positionalDeleteFileReaders_.size());

@@ -61,7 +61,7 @@ class CudfIcebergSplitReader : public CudfSplitReader {
       const std::shared_ptr<io::IoStatistics>& ioStatistics,
       const std::shared_ptr<IoStats>& ioStats,
       bool useExperimentalCudfReader,
-      cudf::ast::expression const* subfieldFilterExpr);
+      SubfieldFilterBuildState subfieldFilterBuildState);
 
   /// Override to setup delete file readers and column projection.
   /// @param runtimeStats Reference to the DataSource's runtime statistics,

@@ -266,8 +266,8 @@ struct GroupbyDecimalSumAggregator : GroupbyAggregator {
   uint32_t countIdx_{0};
   std::unique_ptr<cudf::column> decodedSum_;
   std::unique_ptr<cudf::column> decodedCount_;
-  // Holds the DECIMAL64->DECIMAL128 cast of raw input (kPartial/kSingle), kept
-  // alive while the groupby request references its view.
+  // Holds DECIMAL32/DECIMAL64->DECIMAL128 cast of raw input (kPartial/kSingle),
+  // kept alive while the groupby request references its view.
   std::unique_ptr<cudf::column> castedInput_;
 };
 
@@ -342,8 +342,8 @@ struct GroupbyDecimalAvgAggregator : GroupbyAggregator {
   uint32_t countIdx_{0};
   std::unique_ptr<cudf::column> decodedSum_;
   std::unique_ptr<cudf::column> decodedCount_;
-  // Holds the DECIMAL64->DECIMAL128 cast of raw input (kPartial/kSingle), kept
-  // alive while the groupby request references its view.
+  // Holds DECIMAL32/DECIMAL64->DECIMAL128 cast of raw input (kPartial/kSingle),
+  // kept alive while the groupby request references its view.
   std::unique_ptr<cudf::column> castedInput_;
 };
 

@@ -40,17 +40,54 @@ cudf::column_view castDecimal64InputToDecimal128(
     cudf::column_view inputCol,
     std::unique_ptr<cudf::column>& holder,
     rmm::cuda_stream_view stream) {
-  if (inputCol.type().id() != cudf::type_id::DECIMAL64) {
+  auto inputType = inputCol.type().id();
+  if (inputType == cudf::type_id::DECIMAL128) {
     return inputCol;
   }
-  holder = cudf::cast(
-      inputCol,
-      cudf::data_type{cudf::type_id::DECIMAL128, inputCol.type().scale()},
+  if (inputType == cudf::type_id::DECIMAL32) {
+    holder = cudf::cast(
+        inputCol,
+        cudf::data_type{cudf::type_id::DECIMAL64, inputCol.type().scale()},
+        stream,
+        get_temp_mr());
+    inputCol = holder->view();
+    inputType = inputCol.type().id();
+  }
+  if (inputType == cudf::type_id::DECIMAL64) {
+    holder = cudf::cast(
+        inputCol,
+        cudf::data_type{cudf::type_id::DECIMAL128, inputCol.type().scale()},
+        stream,
+        get_temp_mr());
+    return holder->view();
+  }
+  return inputCol;
+}
+
+namespace {
+
+std::unique_ptr<cudf::column> widenDecimalSumForSerialization(
+    std::unique_ptr<cudf::column> sum,
+    rmm::cuda_stream_view stream,
+    rmm::device_async_resource_ref mr) {
+  const auto sumType = sum->type().id();
+  if (sumType == cudf::type_id::DECIMAL64 ||
+      sumType == cudf::type_id::DECIMAL128) {
+    return sum;
+  }
+  VELOX_CHECK(
+      sumType == cudf::type_id::DECIMAL32,
+      "Unsupported decimal sum column type (type is {})",
+      cudf::type_to_name(sum->type()));
+  return cudf::cast(
+      sum->view(),
+      cudf::data_type{cudf::type_id::DECIMAL64, sum->type().scale()},
       stream,
-      get_temp_mr());
-  return holder->view();
+      mr);
 }
 
+} // namespace
+
 std::unique_ptr<cudf::column> castCountColumnToInt64(
     std::unique_ptr<cudf::column> count,
     rmm::cuda_stream_view stream) {
@@ -67,6 +104,7 @@ std::unique_ptr<cudf::column> serializeDecimalPartialOrIntermediateState(
     rmm::cuda_stream_view stream,
     rmm::device_async_resource_ref mr) {
   count = castCountColumnToInt64(std::move(count), stream);
+  sum = widenDecimalSumForSerialization(std::move(sum), stream, mr);
   return serializeDecimalSumState(sum->view(), count->view(), stream, mr);
 }
 

@@ -38,14 +38,14 @@ namespace facebook::velox::cudf_velox {
 void validateIntermediateColumnType(cudf::column_view const& column);
 
 /**
- * Casts a DECIMAL64 column up to DECIMAL128 (scale preserved) so a subsequent
- * SUM accumulates in 128 bits instead of wrapping. Allocates the casted column
- * from the temporary memory resource into holder and returns its view. Lifetime
- * stays valid only while holder is alive.
+ * Widens DECIMAL32/DECIMAL64 input to DECIMAL128 (scale preserved) so a
+ * subsequent SUM accumulates in 128 bits instead of wrapping. Allocates the
+ * casted column from the temporary memory resource into holder and returns its
+ * view. Lifetime stays valid only while holder is alive.
  *
- * @param inputCol DECIMAL64 input column.
- * @param holder receives ownership of the casted column when inputCol is
- *        DECIMAL64; unchanged otherwise.
+ * @param inputCol DECIMAL32, DECIMAL64, or DECIMAL128 input column.
+ * @param holder receives ownership of the casted column when widening is
+ *        required; unchanged otherwise.
  * @param stream CUDA stream for device work.
  * @return view of inputCol or of the column stored in holder.
  */