Fix row set life cycle in deduplicated column readers when getValues is not called during last read (facebookincubator#275)

Summary:
Pull Request resolved: facebookincubator#275

When the following conditions are satisfied, we got some use-after-free crash in deduplicated column readers:

1. There are at least 2 filters on 2 different columns other than the deduplicated column;
2. One filter is evaluated before the deduplicated column, and some rows survive the filter in first batch;
3. The column reader stored an initial row set as output rows;
4. On second batch, a new run is started;
5. Some rows survive the first filter, but all rows are filtered out after the second filter after deduplicated column is `read()` (but not `getValues()`);
6. On third batch, the column reader should get a larger row set (larger than the initial row set) to trigger a reallocation of row set buffer;
7. In this case, we still pointing to the old row set buffer which is already released, while trying to get the cached value from last run.

Fix this by using an empty row set to retrieve the last cached value; this is ok because the result was all filtered out.

Reviewed By: sdruzkin

Differential Revision: D84204087

fbshipit-source-id: 800fb785b6ce646f0f838f4b600cdf145aafab68

Author: Yuhta

dwio/nimble/velox/selective/VariableLengthColumnReader.cpp

@@ -865,7 +865,7 @@ void DeduplicatedArrayColumnReader::read(
   if (deduplicatedReadHelper_.loadLastRun_ &&
       !deduplicatedReadHelper_.lastRunLoaded_) {
     velox::VectorPtr result;
-    getValues(outputRows(), &result);
+    getValues({}, &result);
   }
   SelectiveListColumnReader::read(offset, rows, incomingNulls);
 }
@@ -882,7 +882,9 @@ void DeduplicatedArrayColumnReader::getValues(
   // next batch in lazy vector scenario.
   auto dictionaryVector = prepareDictionaryArrayResult(
       *result, requestedType_, rows.size(), memoryPool_);
-  setComplexNulls(rows, *result);
+  if (!rows.empty()) {
+    setComplexNulls(rows, *result);
+  }
   auto indices =
       dictionaryVector->mutableIndices(rows.size())->asMutable<vector_size_t>();
   auto* alphabetArray =
@@ -918,8 +920,11 @@ void DeduplicatedArrayColumnReader::getValues(
     lastRunValue_->resize(0);
   }
 
-  auto* nulls = nullsInReadRange_ ? nullsInReadRange_->as<uint64_t>() : nullptr;
-  deduplicatedReadHelper_.populateSelectedIndices(nulls, rows, indices);
+  if (!rows.empty()) {
+    auto* nulls =
+        nullsInReadRange_ ? nullsInReadRange_->as<uint64_t>() : nullptr;
+    deduplicatedReadHelper_.populateSelectedIndices(nulls, rows, indices);
+  }
 }
 
 DeduplicatedMapColumnReader::DeduplicatedMapColumnReader(
@@ -1010,7 +1015,7 @@ void DeduplicatedMapColumnReader::read(
   if (deduplicatedReadHelper_.loadLastRun_ &&
       !deduplicatedReadHelper_.lastRunLoaded_) {
     velox::VectorPtr result;
-    getValues(outputRows(), &result);
+    getValues({}, &result);
   }
   SelectiveMapColumnReader::read(offset, rows, incomingNulls);
 }
@@ -1027,7 +1032,9 @@ void DeduplicatedMapColumnReader::getValues(
   // next batch in lazy vector scenario.
   auto dictionaryVector = prepareDictionaryMapResult(
       *result, requestedType_, rows.size(), memoryPool_);
-  setComplexNulls(rows, *result);
+  if (!rows.empty()) {
+    setComplexNulls(rows, *result);
+  }
   auto indices =
       dictionaryVector->mutableIndices(rows.size())->asMutable<vector_size_t>();
   auto* alphabetMap =
@@ -1073,7 +1080,10 @@ void DeduplicatedMapColumnReader::getValues(
     lastRunValues_->resize(0);
   }
 
-  auto* nulls = nullsInReadRange_ ? nullsInReadRange_->as<uint64_t>() : nullptr;
-  deduplicatedReadHelper_.populateSelectedIndices(nulls, rows, indices);
+  if (!rows.empty()) {
+    auto* nulls =
+        nullsInReadRange_ ? nullsInReadRange_->as<uint64_t>() : nullptr;
+    deduplicatedReadHelper_.populateSelectedIndices(nulls, rows, indices);
+  }
 }
 } // namespace facebook::nimble

dwio/nimble/velox/selective/tests/SelectiveNimbleReaderTest.cpp

@@ -1036,6 +1036,63 @@ TEST_F(SelectiveNimbleReaderTest, arrayWithOffsetsReuseNullResult) {
   validate(*vector, *readers.rowReader, 2, [](auto) { return true; });
 }
 
+TEST_F(SelectiveNimbleReaderTest, arrayWithOffsetsLastRowSetLifeCycle) {
+  std::vector<std::optional<std::vector<std::optional<int64_t>>>> c0, c1, c2;
+  // First batch, one row after filtering, and allocate outputRows_ with a
+  // smaller size.
+  for (int i = 0; i < 16; ++i) {
+    c0.push_back(std::nullopt);
+    c1.push_back({{1}});
+    c2.push_back({{}});
+  }
+  c0.push_back({{}});
+  c1.push_back({{1}});
+  c2.push_back({{}});
+  // Second batch, all filtered out by c2, but c1 reads some value without
+  // calling getValues.
+  c0.push_back(std::nullopt);
+  // Add a null to force setComplexNulls to read last row set before batch 3.
+  c1.push_back(std::nullopt);
+  c2.push_back({{}});
+  for (int i = 0; i < 15; ++i) {
+    c0.push_back(std::nullopt);
+    c1.push_back({{2}});
+    c2.push_back({{}});
+  }
+  c0.push_back({{}});
+  c1.push_back({{2}});
+  c2.push_back(std::nullopt);
+  // Third batch, nothing is filtered out, and force outputRows_ buffer
+  // reallocation.
+  for (int i = 0; i < 17; ++i) {
+    c0.push_back({{}});
+    c1.push_back({{2}});
+    c2.push_back({{}});
+  }
+  auto vector = makeRowVector({
+      makeNullableArrayVector<int64_t>(c0),
+      makeRowVector(
+          {"c1c0"},
+          {makeNullableArrayVector<int64_t>(c1)},
+          // Add some nulls so it is not lazy.
+          [](auto i) { return i == 0; }),
+      makeNullableArrayVector<int64_t>(c2),
+  });
+  VeloxWriterOptions writerOptions;
+  writerOptions.dictionaryArrayColumns = {"c1"};
+  auto fileContent = test::createNimbleFile(*rootPool(), vector, writerOptions);
+  auto scanSpec = std::make_shared<common::ScanSpec>("root");
+  scanSpec->addAllChildFields(*vector->type());
+  scanSpec->childByName("c0")->setFilter(std::make_shared<common::IsNotNull>());
+  scanSpec->childByName("c1")->setFilter(std::make_shared<common::IsNotNull>());
+  scanSpec->childByName("c2")->setFilter(std::make_shared<common::IsNotNull>());
+  scanSpec->disableStatsBasedFilterReorder();
+  auto readers = makeReaders(vector, fileContent, scanSpec);
+  validate(*vector, *readers.rowReader, 17, [](auto i) {
+    return i == 16 || i >= 34;
+  });
+}
+
 TEST_F(SelectiveNimbleReaderTest, slidingWindowMapSubfieldPruning) {
   common::BigintRange keyFilter(2, 2, false);
   checkSlidingWindowMap(