fix: Add missing metadataFilter_ transfer for split prefetch (#15385)

Summary:
Pull Request resolved: #15385

We're seeing crash when prefetch is turned on, ASAN log P2022180855
In `setFromDataSource`:

1.  `scanSpec_ = std::move(source->scanSpec_);` - The scanSpec is moved from old source to new one
2.  The `scanSpec_` contains raw pointers*to Filter objects owned by `source->metadataFilter_`
3.  But `metadataFilter_` is **NOT** transferred from source to this
4.  When `sourceUnique` (the old source) is destroyed at the end of the function, `source->metadataFilter_` is destroyed
5.  The `scanSpec_` in the new HiveDataSource now has **dangling pointers** to the destroyed Filters
6.  Later async operations try to use these pointers → heap-use-after-free

Reviewed By: Yuhta

Differential Revision: D86178967

fbshipit-source-id: 26bdd74a1339f38472ada0ef2308624898c71301

Author: Ke Wang

velox/connectors/hive/HiveDataSource.cpp

@@ -505,6 +505,7 @@ void HiveDataSource::setFromDataSource(
   readerOutputType_ = std::move(source->readerOutputType_);
   source->scanSpec_->moveAdaptationFrom(*scanSpec_);
   scanSpec_ = std::move(source->scanSpec_);
+  metadataFilter_ = std::move(source->metadataFilter_);
   splitReader_ = std::move(source->splitReader_);
   splitReader_->setConnectorQueryCtx(connectorQueryCtx_);
   // New io will be accounted on the stats of 'source'. Add the existing

velox/exec/tests/TableScanTest.cpp

@@ -4318,6 +4318,53 @@ TEST_F(TableScanTest, parallelPrepare) {
       .copyResults(pool_.get());
 }
 
+TEST_F(TableScanTest, parallelPrepareWithSubfieldFilters) {
+  // Test metadataFilter is correctly transferred during split prefetch.
+  constexpr int32_t kNumParallel = 100;
+  auto data = makeRowVector({
+      makeFlatVector<int32_t>(100, [](auto row) { return row; }),
+      makeFlatVector<int64_t>(100, [](auto row) { return row * 2; }),
+  });
+
+  auto filePath = TempFilePath::create();
+  writeToFile(filePath->getPath(), {data});
+
+  auto subfieldFilters = SubfieldFiltersBuilder()
+                             .add("c0", greaterThanOrEqual(10))
+                             .add("c1", lessThan(150))
+                             .build();
+
+  auto outputType = ROW({"c0", "c1"}, {INTEGER(), BIGINT()});
+  auto remainingFilter = parseExpr("c0 % 2 = 0", outputType);
+  auto tableHandle =
+      makeTableHandle(std::move(subfieldFilters), remainingFilter);
+  auto assignments = allRegularColumns(outputType);
+
+  auto plan = exec::test::PlanBuilder(pool_.get())
+                  .startTableScan()
+                  .outputType(outputType)
+                  .tableHandle(tableHandle)
+                  .assignments(assignments)
+                  .endTableScan()
+                  .planNode();
+
+  std::vector<exec::Split> splits;
+  for (auto i = 0; i < kNumParallel; ++i) {
+    splits.push_back(makeHiveSplit(filePath->getPath()));
+  }
+
+  auto result = AssertQueryBuilder(plan)
+                    .config(
+                        core::QueryConfig::kMaxSplitPreloadPerDriver,
+                        std::to_string(kNumParallel))
+                    .splits(splits)
+                    .copyResults(pool_.get());
+
+  // Verify results: c0 >= 10 AND c1 < 150 AND c0 % 2 = 0
+  // So rows [10,12,14,...,74] = 33 rows per split
+  ASSERT_EQ(result->size(), 33 * kNumParallel);
+}
+
 TEST_F(TableScanTest, dictionaryMemo) {
   constexpr int kSize = 100;
   const char* baseStrings[] = {