fix(iceberg): Apply row lineage filters after synthesis

velox/connectors/hive/FileDataSource.cpp

@@ -426,7 +426,24 @@ void FileDataSource::addSplit(std::shared_ptr<ConnectorSplit> split) {
   splitReader_->configureReaderOptions(randomSkip_);
   splitReader_->setRemainingFilterColumns(remainingFilterColumns_);
   splitReader_->prepareSplit(metadataFilter_, runtimeStats_);
-  readerOutputType_ = splitReader_->readerOutputType();
+
+  auto splitReaderOutputType = splitReader_->readerOutputType();
+  if (readerProducedType_ != nullptr &&
+      splitReaderOutputType->size() > readerProducedType_->size()) {
+    // The split reader appended columns, e.g. an unselected Iceberg
+    // equality-delete or lineage column. next() allocates the output from
+    // 'readerProducedType_', so it has to grow by the same columns.
+    auto names = readerProducedType_->names();
+    auto types = readerProducedType_->children();
+    for (auto i = readerProducedType_->size();
+         i < splitReaderOutputType->size();
+         ++i) {
+      names.push_back(splitReaderOutputType->nameOf(i));
+      types.push_back(splitReaderOutputType->childAt(i));
+    }
+    readerProducedType_ = ROW(std::move(names), std::move(types));
+  }
+  readerOutputType_ = std::move(splitReaderOutputType);
 }
 
 std::optional<RowVectorPtr> FileDataSource::next(

velox/connectors/hive/iceberg/EqualityDeleteFileReader.cpp

@@ -339,7 +339,8 @@ void EqualityDeleteFileReader::applyDeletes(
 
   // For each row in the output, compute its hash and probe the delete set.
   for (vector_size_t i = 0; i < output->size(); ++i) {
-    // Skip rows already deleted by positional/DV deletes.
+    // Skip rows already removed by a post-read filter or an earlier delete
+    // file.
     if (bits::isBitSet(bitmap, i)) {
       continue;
     }

velox/connectors/hive/iceberg/EqualityDeleteFileReader.h

@@ -87,8 +87,8 @@ class EqualityDeleteFileReader {
   ///
   /// @param output The base data output vector to filter.
   /// @param deleteBitmap Output bitmap. Bit i is set if row i matches an
-  ///   equality delete. The bitmap must be pre-allocated to cover at least
-  ///   output->size() rows.
+  ///   equality delete. Rows whose bit is already set are skipped. The bitmap
+  ///   must be pre-allocated to cover at least output->size() rows.
   void applyDeletes(const RowVectorPtr& output, BufferPtr deleteBitmap);
 
   /// Returns the number of delete key tuples loaded from the file.