update files

Signed-off-by: tqqq <dirtysalt1987@gmail.com>

Author: dirtysalt

be/src/formats/parquet/column_materializer.h

@@ -229,11 +229,13 @@ class ColumnMaterializer {
     //   1. read_active_range_round_by_round()
     //      - read_slot() reads active columns → PHYSICAL in _read_chunk
     //      - Dict filters run ON physical columns
-    //      - Post-read conjuncts → finalize_lazy_state() → LOGICAL in chunk
+    //      - Post-read conjuncts → finalize_active_slot() → LOGICAL + marks
+    //        _logical_slot_ids so emit bypasses fill_dst_column
     //
-    //   2. Compound predicate eval (ChunkPredicateEvaluator)
+    //   2. Compound predicate eval (GroupReader Phase 2b)
+    //      - finalize_active_slot() on all active columns (also marks
+    //        _logical_slot_ids)
     //      - ColumnRef for missing lazy slots calls MissingColumnProvider
-    //        → LazyMaterializationContext::provide()
     //        → materialize_slot(): read_range() + finalize_lazy_state()
     //        → column in _read_chunk is LOGICAL, cached in _slot_cache
     //

be/src/formats/parquet/group_reader.cpp

@@ -197,8 +197,6 @@ Status GroupReader::get_next(ChunkPtr* chunk, size_t* row_count) {
         return Status::EndOfFile("");
     }
 
-    ChunkPtr active_chunk = _column_materializer->create_active_chunk();
-
     while (true) {
         if (!_range_iter.has_more()) {
             *row_count = 0;
@@ -209,13 +207,19 @@ Status GroupReader::get_next(ChunkPtr* chunk, size_t* row_count) {
         auto count = r.span_size();
         _param.stats->raw_rows_read += count;
 
-        // Previous iteration's triggered-lazy slot cache must not carry over into
-        // the next range — stale slot_cache entries would corrupt read_lazy_columns.
+        // Per-range state reset.  _slot_cache and _logical_slot_ids are
+        // cleared here; _fetched_hidden_slots is cleared by reset_iteration_state.
         _column_materializer->reset_read_chunk();
         _variant->reset_iteration_state();
+
+        // Rebuild active_chunk per range so that slots appended during
+        // predicate evaluation / lazy backfill / variant projection in a
+        // prior range do not leak into the current range.  Chunk::reset()
+        // preserves slot mappings which can leave stale empty columns.
+        ChunkPtr active_chunk = _column_materializer->create_active_chunk();
+
         bool has_filter = false;
         Filter chunk_filter(count, 1);
-        active_chunk->reset();
 
         // Phase 4: lazy materialization context for on-demand slot resolution.
         // Created early so Phase 6 expression trigger can call materialize_slot()
@@ -260,11 +264,9 @@ Status GroupReader::get_next(ChunkPtr* chunk, size_t* row_count) {
         //     variant virtual slots are created by emit_projections() which
         //     runs later.  When a compound conjunct references such a slot,
         //     we fetch only the needed hidden sources and project only the
-        //     needed virtual slots early.  The projected slots stay in
-        //     active_chunk through filtering and are reused by
-        //     emit_projections() at Phase 7.  Any remaining hidden sources
-        //     are fetched by the normal Phase 3 call — fetch_sources()
-        //     skips already-populated slots via _fetched_hidden_slots.
+        //     needed virtual slots early.  Because active_chunk is rebuilt
+        //     per range, there is no risk of stale projected slots from a
+        //     prior range that was fully filtered.
         std::unordered_set<SlotId> early_projected_slots;
         if (!_param.scanner_ctx->conjuncts.scanner_ctxs.empty()) {
             if (!_variant->empty()) {

be/src/formats/parquet/lazy_materialization_context.cpp

@@ -42,11 +42,10 @@ bool LazyMaterializationContext::can_materialize(SlotId slot_id) const {
 }
 
 Status LazyMaterializationContext::materialize_slot(SlotId slot_id) {
-    // Variant lazy hidden source slot: dispatch to VariantProjectionHandler
-    // BEFORE checking has_slot().  Chunk::reset() preserves slot indices
-    // from prior iterations that were fully filtered, so has_slot() returns
-    // true for stale empty columns.  materialize_hidden_source() handles
-    // the stale case by re-reading into update_column.
+    // Variant lazy hidden sources are dispatched to VariantProjectionHandler
+    // before checking has_slot(), because they live outside the regular
+    // slot_cache / slot_id_index.  materialize_hidden_source() own dedup
+    // via _fetched_hidden_slots.
     if (_variant != nullptr) {
         const auto& hidden_lazy = _variant->lazy_hidden_slot_ids();
         if (std::find(hidden_lazy.begin(), hidden_lazy.end(), slot_id) != hidden_lazy.end()) {

be/src/formats/parquet/variant_projection.cpp

@@ -800,10 +800,9 @@ Status VariantProjectionHandler::fetch_and_project_virtual_slots(const std::unor
     }
 
     // 2. Project each needed virtual slot from its source column.
-    //    Re-project every iteration — Chunk::reset() clears column data
-    //    but preserves slot indices (is_slot_exist stays true), so a
-    //    prior range that was fully filtered would leave an empty stale
-    //    column that skipping would reuse without fresh projection.
+    //    active_chunk is rebuilt per range, so append_column is the
+    //    common case.  When called multiple times within the same
+    //    iteration update_column replaces a previously projected slot.
     for (SlotId virtual_slot_id : virtual_slot_ids) {
         auto proj_it = _projections.find(virtual_slot_id);
         DCHECK(proj_it != _projections.end());
@@ -909,11 +908,9 @@ Status VariantProjectionHandler::backfill_sources(const Range<uint64_t>& full_ra
 
     auto lazy_hidden_chunk = std::make_shared<Chunk>();
     for (SlotId slot_id : _lazy_hidden_slot_ids) {
-        // Use _fetched_hidden_slots, not active_chunk->is_slot_exist().
-        // Chunk::reset() preserves slot indices from a prior iteration that
-        // was fully filtered, so is_slot_exist() would return true for stale
-        // empty columns.  _fetched_hidden_slots tracks what was actually
-        // populated THIS iteration.
+        // Skip slots already populated this iteration via on-demand
+        // materialization (materialize_hidden_source) or early fetch
+        // (fetch_and_project_virtual_slots).
         if (_fetched_hidden_slots.count(slot_id)) continue;
 
         auto* hidden_src = _hidden_slot_index.at(slot_id);
@@ -946,23 +943,21 @@ Status VariantProjectionHandler::materialize_hidden_source(SlotId slot_id, const
     if (std::find(lazy_ids.begin(), lazy_ids.end(), slot_id) == lazy_ids.end()) {
         return Status::OK();
     }
-    // Chunk::reset() preserves slot indices but clears data.  A slot may
-    // exist from a prior iteration that was fully filtered.  Re-read the
-    // source in every case — update_column replaces stale data, append_column
-    // is used only on the first population.
+    // Already populated this iteration (by an earlier on-demand trigger).
+    if (_fetched_hidden_slots.count(slot_id)) {
+        return Status::OK();
+    }
     auto it = _hidden_slot_index.find(slot_id);
     if (it == _hidden_slot_index.end()) {
         return Status::InternalError(
                 fmt::format("materialize_hidden_source: slot {} not in hidden_slot_index", slot_id));
     }
+    // active_chunk is rebuilt per range, so the slot never exists from a
+    // prior iteration — always append.
     ColumnPtr col = ColumnHelper::create_column(TypeDescriptor::from_logical_type(TYPE_VARIANT), true);
     RETURN_IF_ERROR(it->second->reader->read_range(range, filter, col));
     RETURN_IF_ERROR(it->second->reader->finalize_lazy_state(col));
-    if (active_chunk->is_slot_exist(slot_id)) {
-        active_chunk->update_column(std::move(col), slot_id);
-    } else {
-        active_chunk->append_column(std::move(col), slot_id);
-    }
+    active_chunk->append_column(std::move(col), slot_id);
     _fetched_hidden_slots.insert(slot_id);
     return Status::OK();
 }

be/src/formats/parquet/variant_projection.h

@@ -131,14 +131,15 @@ class VariantProjectionHandler {
     Status align_after_combined_filter(const ChunkPtr& active_chunk, const Filter& filter, size_t pre_filter_rows);
 
     // 6. Backfill projection-only variant source columns.
-    // Slots already triggered via LazyMaterializationContext (already present in
-    // active_chunk) are skipped to avoid double-reading.
+    // Slots already populated this iteration (tracked by _fetched_hidden_slots)
+    // are skipped to avoid double-reading.
     Status backfill_sources(const Range<uint64_t>& full_range, const Range<uint64_t>* post_filter_range,
                             const Filter* post_filter, bool has_filter, ChunkPtr& active_chunk);
 
     // On-demand materialization of a single lazy hidden source slot into active_chunk.
     // Called by LazyMaterializationContext when an expression triggers the slot.
-    // No-op if slot_id is already in active_chunk or is not a lazy hidden source.
+    // No-op if slot_id is not a lazy hidden source or was already populated this
+    // iteration (tracked by _fetched_hidden_slots).
     Status materialize_hidden_source(SlotId slot_id, const Range<uint64_t>& range, const Filter* filter,
                                      ChunkPtr& active_chunk);