[fix](be) Refresh V2 predicates and bound nested Parquet skips

be/src/exec/scan/file_scanner_v2.cpp

@@ -492,13 +492,19 @@ Status FileScannerV2::_prepare_table_reader_split(const TFileRangeDesc& range,
                                                   std::map<std::string, Field> partition_values) {
     format::FileFormat current_split_format;
     RETURN_IF_ERROR(_to_file_format(get_range_format_type(*_params, range), &current_split_format));
+    VExprContextSPtrs conjuncts;
+    RETURN_IF_ERROR(_build_table_conjuncts(&conjuncts));
     VExprContextSPtrs partition_prune_conjuncts;
     if (_state->query_options().enable_runtime_filter_partition_prune) {
         RETURN_IF_ERROR(_build_table_conjuncts(&partition_prune_conjuncts));
     }
     RETURN_IF_ERROR(_table_reader->prepare_split({
             .partition_values = std::move(partition_values),
+            .conjuncts = std::move(conjuncts),
             .partition_prune_conjuncts = std::move(partition_prune_conjuncts),
+            // A metadata COUNT split may span scheduler turns. Do not enter that irreversible
+            // synthetic-row path while a runtime filter can still arrive between batches.
+            .all_runtime_filters_applied = _applied_rf_num == _total_rf_num,
             .cache = _kv_cache,
             .current_range = range,
             .current_split_format = current_split_format,

be/src/format_v2/jni/jni_table_reader.cpp

@@ -32,14 +32,6 @@ namespace doris::format {
 Status JniTableReader::init(TableReadOptions&& options) {
     RETURN_IF_ERROR(TableReader::init(std::move(options)));
     _init_profile();
-
-    // JNI readers do not go through TableReader::open_reader(), where file-local filters are
-    // prepared for file readers. They execute table-level conjuncts directly on the JNI block.
-    RowDescriptor row_desc;
-    for (const auto& conjunct : _conjuncts) {
-        RETURN_IF_ERROR(conjunct->prepare(_runtime_state, row_desc));
-        RETURN_IF_ERROR(conjunct->open(_runtime_state));
-    }
     return Status::OK();
 }
 
@@ -55,6 +47,13 @@ Status JniTableReader::prepare_split(const SplitReadOptions& options) {
     if (_is_table_level_count_active()) {
         return Status::OK();
     }
+    // JNI readers do not go through TableReader::open_reader(), where native readers prepare
+    // file-local filters. Prepare the fresh per-split snapshot before it filters JNI blocks.
+    RowDescriptor row_desc;
+    for (const auto& conjunct : _conjuncts) {
+        RETURN_IF_ERROR(conjunct->prepare(_runtime_state, row_desc));
+        RETURN_IF_ERROR(conjunct->open(_runtime_state));
+    }
     // Subclasses populate split-specific scanner params before calling this method, so the Java
     // scanner can be opened here instead of being lazily opened by the first get_block() call.
     return _open_jni_scanner();

be/src/format_v2/parquet/reader/column_reader.cpp

@@ -606,14 +606,34 @@ Status ParquetColumnReader::build_nested_column(int64_t, MutableColumnPtr&, int6
                                 _name);
 }
 
-Status ParquetColumnReader::skip_nested_column(int64_t rows) {
-    auto scratch_column = _type->create_column();
-    int64_t values_read = 0;
-    RETURN_IF_ERROR(build_nested_column(rows, scratch_column, &values_read));
-    if (values_read != rows) {
-        return Status::Corruption("Failed to skip nested parquet column {}: skipped {} of {} rows",
-                                  _name, values_read, rows);
+Status ParquetColumnReader::consume_nested_column(int64_t, int64_t*) {
+    return Status::NotSupported("Parquet nested column consume is not supported for column {}",
+                                _name);
+}
+
+Status ParquetColumnReader::skip_nested_rows(int64_t rows) {
+    if (rows <= 0) {
+        return Status::OK();
+    }
+
+    // A nested parent row may expand to many child values. Capping the number of parent rows per
+    // loaded batch bounds that amplification for large holes. The consume interface advances the
+    // loaded definition/repetition levels recursively without constructing a discarded Column.
+    constexpr int64_t MAX_NESTED_SKIP_BATCH_SIZE = 4096;
+    int64_t remaining_rows = rows;
+    while (remaining_rows > 0) {
+        const int64_t batch_rows = std::min(remaining_rows, MAX_NESTED_SKIP_BATCH_SIZE);
+        RETURN_IF_ERROR(load_nested_levels_batch(batch_rows));
+        int64_t rows_consumed = 0;
+        RETURN_IF_ERROR(consume_nested_column(batch_rows, &rows_consumed));
+        if (rows_consumed != batch_rows) {
+            return Status::Corruption(
+                    "Failed to skip nested parquet column {}: skipped {} of {} rows in batch",
+                    _name, rows_consumed, batch_rows);
+        }
+        remaining_rows -= batch_rows;
     }
+    update_reader_skip_rows(rows);
     return Status::OK();
 }
 

be/src/format_v2/parquet/reader/column_reader.h

@@ -80,17 +80,33 @@ class ParquetColumnReader {
 
     virtual Status load_nested_batch(int64_t rows);
 
-    // Shape-only load interface for COUNT(col). Implementations only guarantee that
-    // nested_definition_levels(), nested_repetition_levels(), and nested_levels_written() are available;
-    // value_indices and values_column are not guaranteed, so callers must not call build_nested_column() afterwards.
-    // This protocol lets the V2 aggregation path avoid Doris-side value materialization even when
-    // the representative ARRAY/STRUCT leaf is STRING/BINARY; normal scans still use load_nested_batch().
+    // Shape-only load interface for COUNT(col) and skip. Implementations guarantee only that
+    // nested_definition_levels(), nested_repetition_levels(), and nested_levels_written() are
+    // available; value indices and payload columns may be absent. Callers may inspect the levels or
+    // call consume_nested_column(), but must not call build_nested_column() afterwards. For example,
+    // skipping ARRAY<STRING> uses this method to find ARRAY boundaries without constructing a
+    // ColumnString. The underlying Arrow reader may still decode a page because it has no public
+    // levels-only API. Normal scans that need output values use load_nested_batch() instead.
     virtual Status load_nested_levels_batch(int64_t rows);
 
     virtual Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column,
                                        int64_t* values_read);
 
-    virtual Status skip_nested_column(int64_t rows);
+    // Consume logical values from a batch previously loaded by load_nested_batch() or
+    // load_nested_levels_batch() without appending them to an output Column. Implementations must
+    // advance exactly the same nested level cursors and perform the same shape/null/alignment
+    // validation as build_nested_column(). The levels-only form is preferred for skip paths because
+    // it avoids transferring leaf payloads into Doris Columns when they will be discarded.
+    //
+    // `length_upper_bound` is expressed at this reader's logical level, not in physical leaf
+    // values. For example, consuming two rows from ARRAY [[1, 2], []] consumes two parent ARRAY
+    // rows but only two element values. A MAP implementation must also consume key/value streams
+    // in lockstep, while a nullable STRUCT consumes no child value for a null parent.
+    //
+    // Callers must not use the ordinary skip() after either load call: the leaf stream has already
+    // advanced into an in-memory nested batch, and doing so would advance it twice.
+    // `values_consumed` may be smaller than the requested bound only when the loaded batch ends.
+    virtual Status consume_nested_column(int64_t length_upper_bound, int64_t* values_consumed);
 
     virtual const std::vector<int16_t>& nested_definition_levels() const;
     virtual const std::vector<int16_t>& nested_repetition_levels() const;
@@ -109,6 +125,10 @@ class ParquetColumnReader {
     ParquetColumnReader(const ParquetColumnSchema& schema, const DataTypePtr type,
                         ParquetColumnReaderProfile profile = {});
     ParquetColumnReader() = default;
+    // Load shape levels and consume skipped parent rows in bounded batches. The bound limits level
+    // memory when a parent expands to many children; the levels-only load plus
+    // consume_nested_column() avoids payload materialization and output Columns.
+    Status skip_nested_rows(int64_t rows);
     void update_reader_read_rows(int64_t rows) const;
     void update_reader_skip_rows(int64_t rows) const;
 

be/src/format_v2/parquet/reader/list_column_reader.cpp

@@ -47,19 +47,7 @@ Status ListColumnReader::read(int64_t rows, MutableColumnPtr& column, int64_t* r
 }
 
 Status ListColumnReader::skip(int64_t rows) {
-    if (rows <= 0) {
-        return Status::OK();
-    }
-    auto scratch_column = _type->create_column();
-    RETURN_IF_ERROR(load_nested_batch(rows));
-    int64_t rows_read = 0;
-    RETURN_IF_ERROR(build_nested_column(rows, scratch_column, &rows_read));
-    if (rows_read != rows) {
-        return Status::Corruption("Failed to skip parquet LIST column {}: skipped {} of {} rows",
-                                  _name, rows_read, rows);
-    }
-    update_reader_skip_rows(rows);
-    return Status::OK();
+    return skip_nested_rows(rows);
 }
 
 Status ListColumnReader::load_nested_batch(int64_t rows) {
@@ -68,35 +56,61 @@ Status ListColumnReader::load_nested_batch(int64_t rows) {
     return _element_reader->load_nested_batch(rows);
 }
 
+Status ListColumnReader::load_nested_levels_batch(int64_t rows) {
+    DORIS_CHECK(_element_reader != nullptr);
+    reset_nested_build_level_cursor();
+    return _element_reader->load_nested_levels_batch(rows);
+}
+
 Status ListColumnReader::build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column,
                                              int64_t* values_read) {
-    if (column.get() == nullptr || values_read == nullptr) {
+    if (column.get() == nullptr) {
         return Status::InvalidArgument("Invalid parquet list build result pointer for column {}",
                                        _name);
     }
+    return _consume_or_build_nested_column(length_upper_bound, &column, values_read);
+}
+
+Status ListColumnReader::consume_nested_column(int64_t length_upper_bound,
+                                               int64_t* values_consumed) {
+    return _consume_or_build_nested_column(length_upper_bound, nullptr, values_consumed);
+}
+
+Status ListColumnReader::_consume_or_build_nested_column(int64_t length_upper_bound,
+                                                         MutableColumnPtr* column,
+                                                         int64_t* values_processed) {
+    if (values_processed == nullptr) {
+        return Status::InvalidArgument("Invalid parquet list process result pointer for column {}",
+                                       _name);
+    }
     DORIS_CHECK(_element_reader != nullptr);
-    auto* array_column = array_column_from_output(column);
-    DORIS_CHECK(array_column != nullptr);
-    auto* parent_null_map = null_map_from_nullable_output(column);
-    auto nested_column = array_column->get_data_ptr()->assert_mutable();
-    const auto& element_output_type =
-            assert_cast<const DataTypeArray&>(*remove_nullable(_type)).get_nested_type();
-    remove_nullable_wrapper_if_not_expected(element_output_type, &nested_column);
+    ColumnArray* array_column = nullptr;
+    NullMap* parent_null_map = nullptr;
+    MutableColumnPtr nested_column;
+    if (column != nullptr) {
+        array_column = array_column_from_output(*column);
+        DORIS_CHECK(array_column != nullptr);
+        parent_null_map = null_map_from_nullable_output(*column);
+        nested_column = array_column->get_data_ptr()->assert_mutable();
+        const auto& element_output_type =
+                assert_cast<const DataTypeArray&>(*remove_nullable(_type)).get_nested_type();
+        remove_nullable_wrapper_if_not_expected(element_output_type, &nested_column);
+    }
 
     const auto& def_levels = _element_reader->nested_definition_levels();
     const auto& rep_levels = _element_reader->nested_repetition_levels();
     const int64_t levels_written = _element_reader->nested_levels_written();
     std::vector<uint64_t> entry_counts;
     NullMap parent_nulls;
-    *values_read = 0;
+    *values_processed = 0;
     int64_t level_idx = nested_build_level_cursor();
     const int16_t min_parent_definition_level =
             static_cast<int16_t>(_definition_level - 1 - (_type->is_nullable() ? 1 : 0));
     while (level_idx < levels_written) {
         const int16_t def_level = def_levels[level_idx];
         const int16_t rep_level = rep_levels[level_idx];
         const bool starts_parent = rep_level < _repetition_level;
-        if (starts_parent && *values_read >= length_upper_bound) {
+        if (starts_parent && *values_processed >= length_upper_bound) {
             break;
         }
         ++level_idx;
@@ -116,13 +130,13 @@ Status ListColumnReader::build_nested_column(int64_t length_upper_bound, Mutable
         }
 
         const bool parent_is_null = def_level < _definition_level - 1;
-        if (parent_is_null && parent_null_map == nullptr) {
+        if (parent_is_null && !_type->is_nullable()) {
             return Status::Corruption("Parquet LIST column {} contains null for non-nullable LIST",
                                       _name);
         }
         parent_nulls.push_back(parent_is_null);
         entry_counts.push_back(def_level >= _definition_level ? 1 : 0);
-        ++*values_read;
+        ++*values_processed;
     }
     set_nested_build_level_cursor(level_idx);
 
@@ -132,17 +146,28 @@ Status ListColumnReader::build_nested_column(int64_t length_upper_bound, Mutable
         for (const auto entry_count : entry_counts) {
             total_entries += entry_count;
         }
-        RETURN_IF_ERROR(_element_reader->build_nested_column(static_cast<int64_t>(total_entries),
-                                                             nested_column, &child_value_count));
+        if (column != nullptr) {
+            RETURN_IF_ERROR(_element_reader->build_nested_column(
+                    static_cast<int64_t>(total_entries), nested_column, &child_value_count));
+        } else {
+            RETURN_IF_ERROR(_element_reader->consume_nested_column(
+                    static_cast<int64_t>(total_entries), &child_value_count));
+        }
     } else {
         uint64_t pending_entries = 0;
         auto flush_pending_entries = [&]() -> Status {
             if (pending_entries == 0) {
                 return Status::OK();
             }
             int64_t span_child_value_count = 0;
-            RETURN_IF_ERROR(_element_reader->build_nested_column(
-                    static_cast<int64_t>(pending_entries), nested_column, &span_child_value_count));
+            if (column != nullptr) {
+                RETURN_IF_ERROR(_element_reader->build_nested_column(
+                        static_cast<int64_t>(pending_entries), nested_column,
+                        &span_child_value_count));
+            } else {
+                RETURN_IF_ERROR(_element_reader->consume_nested_column(
+                        static_cast<int64_t>(pending_entries), &span_child_value_count));
+            }
             if (span_child_value_count != static_cast<int64_t>(pending_entries)) {
                 return Status::Corruption(
                         "Parquet LIST column {} built {} child values, expected {}", _name,
@@ -168,9 +193,11 @@ Status ListColumnReader::build_nested_column(int64_t length_upper_bound, Mutable
         return Status::Corruption("Parquet LIST column {} built {} child values, expected {}",
                                   _name, child_value_count, total_entries);
     }
-    array_column->get_data_ptr() = std::move(nested_column);
-    append_offsets(array_column->get_offsets(), entry_counts);
-    append_parent_nulls(parent_null_map, parent_nulls);
+    if (column != nullptr) {
+        array_column->get_data_ptr() = std::move(nested_column);
+        append_offsets(array_column->get_offsets(), entry_counts);
+        append_parent_nulls(parent_null_map, parent_nulls);
+    }
     return Status::OK();
 }
 

be/src/format_v2/parquet/reader/list_column_reader.h

@@ -36,15 +36,20 @@ class ListColumnReader final : public ParquetColumnReader {
     Status read(int64_t rows, MutableColumnPtr& column, int64_t* rows_read) override;
     Status skip(int64_t rows) override;
     Status load_nested_batch(int64_t rows) override;
+    Status load_nested_levels_batch(int64_t rows) override;
     Status build_nested_column(int64_t length_upper_bound, MutableColumnPtr& column,
                                int64_t* values_read) override;
+    Status consume_nested_column(int64_t length_upper_bound, int64_t* values_consumed) override;
     const std::vector<int16_t>& nested_definition_levels() const override;
     const std::vector<int16_t>& nested_repetition_levels() const override;
     int64_t nested_levels_written() const override;
     bool is_or_has_repeated_child() const override;
     void advance_nested_build_level_cursor_past_parent(int16_t parent_repetition_level) override;
 
 private:
+    Status _consume_or_build_nested_column(int64_t length_upper_bound, MutableColumnPtr* column,
+                                           int64_t* values_processed);
+
     std::unique_ptr<ParquetColumnReader>
             _element_reader; // element reader (recursive; may be Scalar/Struct/List/Map)
 };