This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 7047fb2cfa8 branch-4.0: [fix](paimon) preserve partition metadata
across splits (#65583)
7047fb2cfa8 is described below
commit 7047fb2cfa8fc23fe03159ca6b787b5fca06dc9a
Author: Socrates <[email protected]>
AuthorDate: Wed Jul 15 17:55:46 2026 +0800
branch-4.0: [fix](paimon) preserve partition metadata across splits (#65583)
### What problem does this PR solve?
Paimon partition metadata was generated only when runtime partition
pruning was enabled. Some count/native/JNI splits therefore reached BE
without stable partition values. BE also reused a partition-prune block
across splits but inserted replacement columns, corrupting the block
layout.
### Changes
- Always attach ordered Paimon partition metadata to every data split.
- Generate aligned `columns_from_path_keys`, `columns_from_path`, and
`columns_from_path_is_null` arrays in FE.
- Pass explicit null markers from `FileScanner` to Parquet/ORC readers.
- Preserve NULL, empty string, and literal `\N` semantics.
- Replace partition columns in the reusable prune block instead of
inserting new columns.
- Preserve the partition-key case behavior already present on the latest
branch-4.0 baseline.
- Add FE/BE unit coverage and a three-partition static `IN` regression
case.
This is a focused backport of the relevant #62821 behavior adapted to
the pre-#62306 reader architecture.
---
be/src/vec/exec/format/generic_reader.h | 3 +-
be/src/vec/exec/format/orc/vorc_reader.cpp | 28 +++----
be/src/vec/exec/format/orc/vorc_reader.h | 4 +-
.../exec/format/parquet/vparquet_group_reader.cpp | 24 ++----
.../exec/format/parquet/vparquet_group_reader.h | 3 +-
be/src/vec/exec/format/parquet/vparquet_reader.cpp | 4 +-
be/src/vec/exec/format/parquet/vparquet_reader.h | 3 +-
.../exec/format/table/partition_column_filler.h | 79 +++++++++++++++++++
be/src/vec/exec/format/table/table_format_reader.h | 6 +-
be/src/vec/exec/scan/file_scanner.cpp | 79 ++++++-------------
.../format/table/partition_column_filler_test.cpp | 88 ++++++++++++++++++++++
.../apache/doris/datasource/paimon/PaimonUtil.java | 10 +++
.../datasource/paimon/source/PaimonScanNode.java | 60 +++++++++------
.../datasource/paimon/source/PaimonSplit.java | 6 +-
.../doris/datasource/paimon/PaimonUtilTest.java | 32 ++++++++
.../paimon/source/PaimonScanNodeTest.java | 29 ++++++-
...est_paimon_runtime_filter_partition_pruning.out | 7 +-
..._paimon_runtime_filter_partition_pruning.groovy | 6 +-
18 files changed, 346 insertions(+), 125 deletions(-)
diff --git a/be/src/vec/exec/format/generic_reader.h
b/be/src/vec/exec/format/generic_reader.h
index 620112a71e7..71893fa994f 100644
--- a/be/src/vec/exec/format/generic_reader.h
+++ b/be/src/vec/exec/format/generic_reader.h
@@ -75,7 +75,8 @@ public:
virtual Status set_fill_columns(
const std::unordered_map<std::string, std::tuple<std::string,
const SlotDescriptor*>>&
partition_columns,
- const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns) {
+ const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns,
+ const std::unordered_map<std::string, bool>&
partition_value_is_null = {}) {
return Status::OK();
}
diff --git a/be/src/vec/exec/format/orc/vorc_reader.cpp
b/be/src/vec/exec/format/orc/vorc_reader.cpp
index 55913981255..dd83dbb599a 100644
--- a/be/src/vec/exec/format/orc/vorc_reader.cpp
+++ b/be/src/vec/exec/format/orc/vorc_reader.cpp
@@ -84,6 +84,7 @@
#include "vec/data_types/data_type_nullable.h"
#include "vec/data_types/data_type_struct.h"
#include "vec/exec/format/orc/orc_file_reader.h"
+#include "vec/exec/format/table/partition_column_filler.h"
#include "vec/exec/format/table/transactional_hive_common.h"
#include "vec/exec/scan/file_scanner.h"
#include "vec/exprs/vbloom_predicate.h"
@@ -1064,8 +1065,10 @@ bool OrcReader::_init_search_argument(const VExprSPtrs&
exprs) {
Status OrcReader::set_fill_columns(
const std::unordered_map<std::string, std::tuple<std::string, const
SlotDescriptor*>>&
partition_columns,
- const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns) {
+ const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns,
+ const std::unordered_map<std::string, bool>& partition_value_is_null) {
SCOPED_RAW_TIMER(&_statistics.set_fill_column_time);
+ _lazy_read_ctx.partition_value_is_null = partition_value_is_null;
// std::unordered_map<column_name, std::pair<col_id, slot_id>>
std::unordered_map<std::string, std::pair<uint32_t, int>>
predicate_table_columns;
@@ -1383,23 +1386,12 @@ Status OrcReader::_fill_partition_columns(
auto col_ptr =
block->get_by_position((*_col_name_to_block_idx)[kv.first])
.column->assume_mutable();
const auto& [value, slot_desc] = kv.second;
- auto _text_serde = slot_desc->get_data_type_ptr()->get_serde();
- Slice slice(value.data(), value.size());
- uint64_t num_deserialized = 0;
- if (_text_serde->deserialize_column_from_fixed_json(*col_ptr, slice,
rows,
- &num_deserialized,
-
_text_formatOptions) != Status::OK()) {
- return Status::InternalError("Failed to fill partition column:
{}={}",
- slot_desc->col_name(), value);
- }
- if (num_deserialized != rows) {
- return Status::InternalError(
- "Failed to fill partition column: {}={} ."
- "Number of rows expected to be written : {}, number of
rows actually "
- "written : "
- "{}",
- slot_desc->col_name(), value, num_deserialized, rows);
- }
+ auto null_it = _lazy_read_ctx.partition_value_is_null.find(kv.first);
+ RETURN_IF_ERROR(fill_partition_column_from_path_value(
+ *col_ptr, *slot_desc, value, rows,
+ null_it != _lazy_read_ctx.partition_value_is_null.end(),
+ null_it != _lazy_read_ctx.partition_value_is_null.end() &&
null_it->second,
+ _text_formatOptions));
}
return Status::OK();
}
diff --git a/be/src/vec/exec/format/orc/vorc_reader.h
b/be/src/vec/exec/format/orc/vorc_reader.h
index ef2eb7976b4..418c326664b 100644
--- a/be/src/vec/exec/format/orc/vorc_reader.h
+++ b/be/src/vec/exec/format/orc/vorc_reader.h
@@ -109,6 +109,7 @@ struct LazyReadContext {
// lazy read partition columns or all partition columns
std::unordered_map<std::string, std::tuple<std::string, const
SlotDescriptor*>>
partition_columns;
+ std::unordered_map<std::string, bool> partition_value_is_null;
std::unordered_map<std::string, VExprContextSPtr>
predicate_missing_columns;
// lazy read missing columns or all missing columns
std::unordered_map<std::string, VExprContextSPtr> missing_columns;
@@ -180,7 +181,8 @@ public:
Status set_fill_columns(
const std::unordered_map<std::string, std::tuple<std::string,
const SlotDescriptor*>>&
partition_columns,
- const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns) override;
+ const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns,
+ const std::unordered_map<std::string, bool>&
partition_value_is_null = {}) override;
Status get_next_block(Block* block, size_t* read_rows, bool* eof) override;
diff --git a/be/src/vec/exec/format/parquet/vparquet_group_reader.cpp
b/be/src/vec/exec/format/parquet/vparquet_group_reader.cpp
index 2cec1061218..95f6c13a575 100644
--- a/be/src/vec/exec/format/parquet/vparquet_group_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_group_reader.cpp
@@ -54,6 +54,7 @@
#include "vec/data_types/data_type_nullable.h"
#include "vec/data_types/data_type_number.h"
#include "vec/data_types/data_type_string.h"
+#include "vec/exec/format/table/partition_column_filler.h"
#include "vec/exprs/vdirect_in_predicate.h"
#include "vec/exprs/vectorized_fn_call.h"
#include "vec/exprs/vexpr.h"
@@ -725,23 +726,12 @@ Status RowGroupReader::_fill_partition_columns(
// obtained from block*, it is a mutable object.
auto* col_ptr = const_cast<IColumn*>(doris_column.get());
const auto& [value, slot_desc] = kv.second;
- auto _text_serde = slot_desc->get_data_type_ptr()->get_serde();
- Slice slice(value.data(), value.size());
- uint64_t num_deserialized = 0;
- // Be careful when reading empty rows from parquet row groups.
- if (_text_serde->deserialize_column_from_fixed_json(*col_ptr, slice,
rows,
- &num_deserialized,
-
_text_formatOptions) != Status::OK()) {
- return Status::InternalError("Failed to fill partition column:
{}={}",
- slot_desc->col_name(), value);
- }
- if (num_deserialized != rows) {
- return Status::InternalError(
- "Failed to fill partition column: {}={} ."
- "Number of rows expected to be written : {}, number of
rows actually written : "
- "{}",
- slot_desc->col_name(), value, num_deserialized, rows);
- }
+ auto null_it = _lazy_read_ctx.partition_value_is_null.find(kv.first);
+ RETURN_IF_ERROR(fill_partition_column_from_path_value(
+ *col_ptr, *slot_desc, value, rows,
+ null_it != _lazy_read_ctx.partition_value_is_null.end(),
+ null_it != _lazy_read_ctx.partition_value_is_null.end() &&
null_it->second,
+ _text_formatOptions));
}
return Status::OK();
}
diff --git a/be/src/vec/exec/format/parquet/vparquet_group_reader.h
b/be/src/vec/exec/format/parquet/vparquet_group_reader.h
index 1e5259413a3..dab92d669e9 100644
--- a/be/src/vec/exec/format/parquet/vparquet_group_reader.h
+++ b/be/src/vec/exec/format/parquet/vparquet_group_reader.h
@@ -82,9 +82,10 @@ public:
// all conjuncts: in sql, join runtime filter, topn runtime filter.
VExprContextSPtrs conjuncts;
- // ParquetReader::set_fill_columns(xxx, xxx) will set these two members
+ // ParquetReader::set_fill_columns(xxx, xxx) will set these members
std::unordered_map<std::string, std::tuple<std::string, const
SlotDescriptor*>>
fill_partition_columns;
+ std::unordered_map<std::string, bool> partition_value_is_null;
std::unordered_map<std::string, VExprContextSPtr> fill_missing_columns;
phmap::flat_hash_map<int,
std::vector<std::shared_ptr<ColumnPredicate>>>
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.cpp
b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
index af1958aa7a1..e559aa43871 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
@@ -464,8 +464,10 @@ bool ParquetReader::_type_matches(const int cid) const {
Status ParquetReader::set_fill_columns(
const std::unordered_map<std::string, std::tuple<std::string, const
SlotDescriptor*>>&
partition_columns,
- const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns) {
+ const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns,
+ const std::unordered_map<std::string, bool>& partition_value_is_null) {
_lazy_read_ctx.fill_partition_columns = partition_columns;
+ _lazy_read_ctx.partition_value_is_null = partition_value_is_null;
_lazy_read_ctx.fill_missing_columns = missing_columns;
// std::unordered_map<column_name, std::pair<col_id, slot_id>>
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.h
b/be/src/vec/exec/format/parquet/vparquet_reader.h
index c67aab46b69..537691ff9ed 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.h
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.h
@@ -164,7 +164,8 @@ public:
Status set_fill_columns(
const std::unordered_map<std::string, std::tuple<std::string,
const SlotDescriptor*>>&
partition_columns,
- const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns) override;
+ const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns,
+ const std::unordered_map<std::string, bool>&
partition_value_is_null = {}) override;
Status get_file_metadata_schema(const FieldDescriptor** ptr);
diff --git a/be/src/vec/exec/format/table/partition_column_filler.h
b/be/src/vec/exec/format/table/partition_column_filler.h
new file mode 100644
index 00000000000..48e4830ff1c
--- /dev/null
+++ b/be/src/vec/exec/format/table/partition_column_filler.h
@@ -0,0 +1,79 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <glog/logging.h>
+
+#include <string>
+
+#include "common/status.h"
+#include "runtime/descriptors.h"
+#include "util/slice.h"
+#include "vec/columns/column_nullable.h"
+#include "vec/common/assert_cast.h"
+#include "vec/data_types/serde/data_type_serde.h"
+
+namespace doris::vectorized {
+
+inline Status fill_partition_column_from_path_value(
+ IColumn& column, const SlotDescriptor& slot_desc, const std::string&
value, size_t rows,
+ bool has_explicit_null_marker, bool explicit_null_marker,
+ DataTypeSerDe::FormatOptions text_format_options = {}) {
+ auto data_type = slot_desc.get_data_type_ptr();
+ auto text_serde = data_type->get_serde();
+ uint64_t num_deserialized = 0;
+
+ if (has_explicit_null_marker && explicit_null_marker) {
+ DCHECK(data_type->is_nullable());
+ column.insert_many_defaults(rows);
+ return Status::OK();
+ }
+
+ IColumn* value_column = &column;
+ DataTypeSerDeSPtr value_serde = text_serde;
+ ColumnNullable* nullable_column = nullptr;
+ // Legacy path partitions encode null as "\\N" and rely on nullable serde.
Sources with an
+ // explicit null marker deserialize the nested value so a literal "\\N"
remains non-null.
+ if (data_type->is_nullable() && has_explicit_null_marker) {
+ nullable_column = assert_cast<ColumnNullable*>(&column);
+ value_column = &nullable_column->get_nested_column();
+ value_serde = text_serde->get_nested_serdes()[0];
+ }
+
+ const size_t old_size = value_column->size();
+ Slice slice(value.data(), value.size());
+ Status status = value_serde->deserialize_column_from_fixed_json(
+ *value_column, slice, rows, &num_deserialized,
text_format_options);
+ if (!status.ok()) {
+ value_column->resize(old_size);
+ return Status::InternalError("Failed to fill partition column: {}={}",
slot_desc.col_name(),
+ value);
+ }
+ if (num_deserialized != rows) {
+ value_column->resize(old_size);
+ return Status::InternalError(
+ "Failed to fill partition column: {}={}. Expected rows: {},
actual: {}",
+ slot_desc.col_name(), value, rows, num_deserialized);
+ }
+ if (nullable_column != nullptr) {
+ nullable_column->get_null_map_column().insert_many_vals(0, rows);
+ }
+ return Status::OK();
+}
+
+} // namespace doris::vectorized
diff --git a/be/src/vec/exec/format/table/table_format_reader.h
b/be/src/vec/exec/format/table/table_format_reader.h
index d32170c0afa..c980d754ca1 100644
--- a/be/src/vec/exec/format/table/table_format_reader.h
+++ b/be/src/vec/exec/format/table/table_format_reader.h
@@ -96,8 +96,10 @@ public:
Status set_fill_columns(
const std::unordered_map<std::string, std::tuple<std::string,
const SlotDescriptor*>>&
partition_columns,
- const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns) final {
- return _file_format_reader->set_fill_columns(partition_columns,
missing_columns);
+ const std::unordered_map<std::string, VExprContextSPtr>&
missing_columns,
+ const std::unordered_map<std::string, bool>&
partition_value_is_null = {}) final {
+ return _file_format_reader->set_fill_columns(partition_columns,
missing_columns,
+ partition_value_is_null);
}
bool fill_all_columns() const override { return
_file_format_reader->fill_all_columns(); }
diff --git a/be/src/vec/exec/scan/file_scanner.cpp
b/be/src/vec/exec/scan/file_scanner.cpp
index d164112c6a9..1a770a002b9 100644
--- a/be/src/vec/exec/scan/file_scanner.cpp
+++ b/be/src/vec/exec/scan/file_scanner.cpp
@@ -70,6 +70,7 @@
#include "vec/exec/format/table/max_compute_jni_reader.h"
#include "vec/exec/format/table/paimon_jni_reader.h"
#include "vec/exec/format/table/paimon_reader.h"
+#include "vec/exec/format/table/partition_column_filler.h"
#include "vec/exec/format/table/remote_doris_reader.h"
#include "vec/exec/format/table/transactional_hive_reader.h"
#include "vec/exec/format/table/trino_connector_jni_reader.h"
@@ -253,33 +254,12 @@ Status
FileScanner::_process_runtime_filters_partition_prune(bool& can_filter_al
for (auto const& partition_col_desc : _partition_col_descs) {
const auto& [partition_value, partition_slot_desc] =
partition_col_desc.second;
auto data_type = partition_slot_desc->get_data_type_ptr();
- auto test_serde = data_type->get_serde();
auto partition_value_column = data_type->create_column();
- auto* col_ptr = static_cast<IColumn*>(partition_value_column.get());
- Slice slice(partition_value.data(), partition_value.size());
- uint64_t num_deserialized = 0;
- DataTypeSerDe::FormatOptions options {};
- if
(_partition_value_is_null.contains(partition_slot_desc->col_name())) {
- // for iceberg/paimon table
- // NOTICE: column is always be nullable for iceberg/paimon table
now
- DCHECK(data_type->is_nullable());
- test_serde = test_serde->get_nested_serdes()[0];
- auto* null_column = assert_cast<ColumnNullable*>(col_ptr);
- if (_partition_value_is_null[partition_slot_desc->col_name()]) {
- null_column->insert_many_defaults(partition_value_column_size);
- } else {
- // If the partition value is not null, we set null map to 0
and deserialize it normally.
- null_column->get_null_map_column().insert_many_vals(0,
partition_value_column_size);
- RETURN_IF_ERROR(test_serde->deserialize_column_from_fixed_json(
- null_column->get_nested_column(), slice,
partition_value_column_size,
- &num_deserialized, options));
- }
- } else {
- // for hive/hudi table, the null value is set as "\\N"
- // TODO: this will be unified as iceberg/paimon table in the future
- RETURN_IF_ERROR(test_serde->deserialize_column_from_fixed_json(
- *col_ptr, slice, partition_value_column_size,
&num_deserialized, options));
- }
+ auto null_it =
_partition_value_is_null.find(partition_slot_desc->col_name());
+ RETURN_IF_ERROR(fill_partition_column_from_path_value(
+ *partition_value_column, *partition_slot_desc, partition_value,
+ partition_value_column_size, null_it !=
_partition_value_is_null.end(),
+ null_it != _partition_value_is_null.end() && null_it->second));
partition_slot_id_to_column[partition_slot_desc->id()] =
std::move(partition_value_column);
}
@@ -291,20 +271,9 @@ Status
FileScanner::_process_runtime_filters_partition_prune(bool& can_filter_al
for (auto const* slot_desc : _real_tuple_desc->slots()) {
if (partition_slot_id_to_column.find(slot_desc->id()) !=
partition_slot_id_to_column.end()) {
- auto data_type = slot_desc->get_data_type_ptr();
auto partition_value_column =
std::move(partition_slot_id_to_column[slot_desc->id()]);
- if (data_type->is_nullable()) {
- _runtime_filter_partition_prune_block.insert(
- index, ColumnWithTypeAndName(
- ColumnNullable::create(
-
std::move(partition_value_column),
-
ColumnUInt8::create(partition_value_column_size, 0)),
- data_type, slot_desc->col_name()));
- } else {
- _runtime_filter_partition_prune_block.insert(
- index,
ColumnWithTypeAndName(std::move(partition_value_column), data_type,
- slot_desc->col_name()));
- }
+ _runtime_filter_partition_prune_block.replace_by_position(
+ index, std::move(partition_value_column));
if (index == 0) {
first_column_filled = true;
}
@@ -638,22 +607,10 @@ Status FileScanner::_fill_columns_from_path(size_t rows) {
// _src_block_ptr points to a mutable block created by this class
itself, so const_cast can be used here.
IColumn* col_ptr = const_cast<IColumn*>(doris_column.get());
auto& [value, slot_desc] = kv.second;
- auto _text_serde = slot_desc->get_data_type_ptr()->get_serde();
- Slice slice(value.data(), value.size());
- uint64_t num_deserialized = 0;
- if (_text_serde->deserialize_column_from_fixed_json(*col_ptr, slice,
rows,
- &num_deserialized,
-
_text_formatOptions) != Status::OK()) {
- return Status::InternalError("Failed to fill partition column:
{}={}",
- slot_desc->col_name(), value);
- }
- if (num_deserialized != rows) {
- return Status::InternalError(
- "Failed to fill partition column: {}={} ."
- "Number of rows expected to be written : {}, number of
rows actually written : "
- "{}",
- slot_desc->col_name(), value, num_deserialized, rows);
- }
+ auto null_it = _partition_value_is_null.find(kv.first);
+ RETURN_IF_ERROR(fill_partition_column_from_path_value(
+ *col_ptr, *slot_desc, value, rows, null_it !=
_partition_value_is_null.end(),
+ null_it != _partition_value_is_null.end() && null_it->second,
_text_formatOptions));
}
return Status::OK();
}
@@ -1440,7 +1397,8 @@ Status FileScanner::_set_fill_or_truncate_columns(bool
need_to_get_parsed_schema
RETURN_IF_ERROR(_generate_missing_columns());
if (_fill_partition_from_path) {
- RETURN_IF_ERROR(_cur_reader->set_fill_columns(_partition_col_descs,
_missing_col_descs));
+ RETURN_IF_ERROR(_cur_reader->set_fill_columns(_partition_col_descs,
_missing_col_descs,
+
_partition_value_is_null));
} else {
// If the partition columns are not from path, we only fill the
missing columns.
RETURN_IF_ERROR(_cur_reader->set_fill_columns({}, _missing_col_descs));
@@ -1578,6 +1536,9 @@ Status FileScanner::_generate_partition_columns() {
_partition_value_is_null.clear();
const TFileRangeDesc& range = _current_range;
if (range.__isset.columns_from_path && !_partition_slot_descs.empty()) {
+ if (range.__isset.columns_from_path_is_null) {
+ DORIS_CHECK(range.columns_from_path_is_null.size() ==
range.columns_from_path.size());
+ }
for (const auto& slot_desc : _partition_slot_descs) {
if (slot_desc) {
auto it = _partition_slot_index_map.find(slot_desc->id());
@@ -1585,6 +1546,12 @@ Status FileScanner::_generate_partition_columns() {
return Status::InternalError("Unknown source slot
descriptor, slot_id={}",
slot_desc->id());
}
+ if (it->second < 0 ||
+ static_cast<size_t>(it->second) >=
range.columns_from_path.size()) {
+ return Status::InternalError(
+ "Invalid partition value index {}, value count {}
for column {}",
+ it->second, range.columns_from_path.size(),
slot_desc->col_name());
+ }
const std::string& column_from_path =
range.columns_from_path[it->second];
_partition_col_descs.emplace(slot_desc->col_name(),
std::make_tuple(column_from_path,
slot_desc));
diff --git a/be/test/vec/exec/format/table/partition_column_filler_test.cpp
b/be/test/vec/exec/format/table/partition_column_filler_test.cpp
new file mode 100644
index 00000000000..130a78c54bf
--- /dev/null
+++ b/be/test/vec/exec/format/table/partition_column_filler_test.cpp
@@ -0,0 +1,88 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "vec/exec/format/table/partition_column_filler.h"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+
+#include "vec/columns/column_nullable.h"
+#include "vec/common/assert_cast.h"
+#include "vec/data_types/data_type_nullable.h"
+#include "vec/data_types/data_type_number.h"
+#include "vec/data_types/data_type_string.h"
+
+namespace doris::vectorized {
+
+TEST(PartitionColumnFillerTest, DistinguishLegacyAndExplicitNullMarkers) {
+ SlotDescriptor string_slot;
+ string_slot._type = make_nullable(std::make_shared<DataTypeString>());
+ string_slot._col_name = "part_col";
+
+ auto legacy_null_column = string_slot.get_empty_mutable_column();
+ ASSERT_TRUE(fill_partition_column_from_path_value(*legacy_null_column,
string_slot, "\\N", 2,
+ false, false)
+ .ok());
+ const auto& legacy_nullable = assert_cast<const
ColumnNullable&>(*legacy_null_column);
+ ASSERT_EQ(legacy_nullable.size(), 2);
+ EXPECT_TRUE(legacy_nullable.is_null_at(0));
+ EXPECT_TRUE(legacy_nullable.is_null_at(1));
+
+ auto literal_null_marker_column = string_slot.get_empty_mutable_column();
+
ASSERT_TRUE(fill_partition_column_from_path_value(*literal_null_marker_column,
string_slot,
+ "\\N", 1, true, false)
+ .ok());
+ const auto& literal_nullable = assert_cast<const
ColumnNullable&>(*literal_null_marker_column);
+ ASSERT_EQ(literal_nullable.size(), 1);
+ EXPECT_FALSE(literal_nullable.is_null_at(0));
+ EXPECT_EQ(literal_nullable.get_nested_column().get_data_at(0).to_string(),
"\\N");
+
+ auto empty_string_column = string_slot.get_empty_mutable_column();
+ ASSERT_TRUE(fill_partition_column_from_path_value(*empty_string_column,
string_slot, "", 1,
+ true, false)
+ .ok());
+ const auto& empty_string_nullable = assert_cast<const
ColumnNullable&>(*empty_string_column);
+ ASSERT_EQ(empty_string_nullable.size(), 1);
+ EXPECT_FALSE(empty_string_nullable.is_null_at(0));
+
EXPECT_EQ(empty_string_nullable.get_nested_column().get_data_at(0).to_string(),
"");
+
+ SlotDescriptor int_slot;
+ int_slot._type = make_nullable(std::make_shared<DataTypeInt32>());
+ int_slot._col_name = "int_part_col";
+ auto explicit_null_column = int_slot.get_empty_mutable_column();
+ ASSERT_TRUE(fill_partition_column_from_path_value(*explicit_null_column,
int_slot, "", 1, true,
+ true)
+ .ok());
+ const auto& explicit_nullable = assert_cast<const
ColumnNullable&>(*explicit_null_column);
+ ASSERT_EQ(explicit_nullable.size(), 1);
+ EXPECT_TRUE(explicit_nullable.is_null_at(0));
+}
+
+TEST(PartitionColumnFillerTest, RestoreColumnAfterDeserializeFailure) {
+ SlotDescriptor int_slot;
+ int_slot._type = std::make_shared<DataTypeInt32>();
+ int_slot._col_name = "int_part_col";
+ auto column = int_slot.get_empty_mutable_column();
+
+ auto status =
+ fill_partition_column_from_path_value(*column, int_slot,
"not_an_int", 1, false, false);
+ EXPECT_FALSE(status.ok());
+ EXPECT_EQ(column->size(), 0);
+}
+
+} // namespace doris::vectorized
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java
index 3239035f20d..3a87bb05274 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/PaimonUtil.java
@@ -487,6 +487,16 @@ public class PaimonUtil {
return null;
}
return value.toString();
+ case FLOAT:
+ if (value == null) {
+ return null;
+ }
+ return Float.toString((Float) value);
+ case DOUBLE:
+ if (value == null) {
+ return null;
+ }
+ return Double.toString((Double) value);
// case binary:
// case varbinary: should not supported, because if return string
with utf8,
// the data maybe be corrupted
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
index 5b24ddfccde..d93e6f58323 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java
@@ -51,6 +51,7 @@ import org.apache.doris.thrift.TPushAggOp;
import org.apache.doris.thrift.TTableFormatFileDesc;
import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.paimon.data.BinaryRow;
@@ -228,6 +229,40 @@ public class PaimonScanNode extends FileQueryScanNode {
}
}
+ private List<String> getOrderedPathPartitionKeys() {
+ if (source == null) {
+ return Collections.emptyList();
+ }
+ return source.getPaimonTable().partitionKeys();
+ }
+
+ @VisibleForTesting
+ void setPartitionValues(TFileRangeDesc rangeDesc, Map<String, String>
partitionValues) {
+ rangeDesc.unsetColumnsFromPathKeys();
+ rangeDesc.unsetColumnsFromPath();
+ rangeDesc.unsetColumnsFromPathIsNull();
+
+ List<String> orderedPartitionKeys = getOrderedPathPartitionKeys();
+ if (orderedPartitionKeys.isEmpty()) {
+ return;
+ }
+ Preconditions.checkState(partitionValues != null,
+ "Missing partition values for Paimon partitioned table");
+
+ List<String> fromPathValues = new
ArrayList<>(orderedPartitionKeys.size());
+ List<Boolean> fromPathIsNull = new
ArrayList<>(orderedPartitionKeys.size());
+ for (String partitionKey : orderedPartitionKeys) {
+ Preconditions.checkState(partitionValues.containsKey(partitionKey),
+ "Missing partition value for Paimon partition key: %s",
partitionKey);
+ String partitionValue = partitionValues.get(partitionKey);
+ fromPathValues.add(partitionValue == null ? "" : partitionValue);
+ fromPathIsNull.add(partitionValue == null);
+ }
+ rangeDesc.setColumnsFromPathKeys(orderedPartitionKeys);
+ rangeDesc.setColumnsFromPath(fromPathValues);
+ rangeDesc.setColumnsFromPathIsNull(fromPathIsNull);
+ }
+
private void setPaimonParams(TFileRangeDesc rangeDesc, PaimonSplit
paimonSplit) {
TTableFormatFileDesc tableFormatFileDesc = new TTableFormatFileDesc();
tableFormatFileDesc.setTableFormatType(paimonSplit.getTableFormatType().value());
@@ -278,20 +313,7 @@ public class PaimonScanNode extends FileQueryScanNode {
fileDesc.setPaimonOptions(backendPaimonOptions);
}
tableFormatFileDesc.setPaimonParams(fileDesc);
- Map<String, String> partitionValues =
paimonSplit.getPaimonPartitionValues();
- if (partitionValues != null) {
- List<String> fromPathKeys = new ArrayList<>();
- List<String> fromPathValues = new ArrayList<>();
- List<Boolean> fromPathIsNull = new ArrayList<>();
- for (Map.Entry<String, String> entry : partitionValues.entrySet())
{
- fromPathKeys.add(entry.getKey());
- fromPathValues.add(entry.getValue() != null ? entry.getValue()
: "");
- fromPathIsNull.add(entry.getValue() == null);
- }
- rangeDesc.setColumnsFromPathKeys(fromPathKeys);
- rangeDesc.setColumnsFromPath(fromPathValues);
- rangeDesc.setColumnsFromPathIsNull(fromPathIsNull);
- }
+ setPartitionValues(rangeDesc, paimonSplit.getPaimonPartitionValues());
rangeDesc.setTableFormatParams(tableFormatFileDesc);
}
@@ -340,6 +362,7 @@ public class PaimonScanNode extends FileQueryScanNode {
// partition data.
// And for counting the number of selected partitions for this paimon
table.
Map<BinaryRow, Map<String, String>> partitionInfoMaps = new
HashMap<>();
+ boolean needPartitionMetadata =
!getOrderedPathPartitionKeys().isEmpty();
// if applyCountPushdown is true, we can't split the DataSplit
boolean hasDeterminedTargetFileSplitSize = false;
long targetFileSplitSize = 0;
@@ -349,9 +372,7 @@ public class PaimonScanNode extends FileQueryScanNode {
BinaryRow partitionValue = dataSplit.partition();
Map<String, String> partitionInfoMap = null;
- if (sessionVariable.isEnableRuntimeFilterPartitionPrune()) {
- // If the partition value is not in the map, we need to
calculate the partition
- // info map and store it in the map.
+ if (needPartitionMetadata) {
partitionInfoMap =
partitionInfoMaps.computeIfAbsent(partitionValue, k -> {
return PaimonUtil.getPartitionInfoMap(
source.getPaimonTable(), partitionValue,
sessionVariable.getTimeZone());
@@ -593,10 +614,7 @@ public class PaimonScanNode extends FileQueryScanNode {
@Override
public List<String> getPathPartitionKeys() throws DdlException,
MetaNotFoundException {
- // return new ArrayList<>(source.getPaimonTable().partitionKeys());
- // Paimon is not aware of partitions and bypasses some existing logic
by
- // returning an empty list
- return new ArrayList<>();
+ return getOrderedPathPartitionKeys();
}
@Override
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonSplit.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonSplit.java
index 00700c3827f..52c2fa8e7e9 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonSplit.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonSplit.java
@@ -26,6 +26,7 @@ import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.DeletionFile;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -41,7 +42,7 @@ public class PaimonSplit extends FileSplit {
private Map<String, String> paimonPartitionValues = null;
public PaimonSplit(DataSplit split) {
- super(DUMMY_PATH, 0, 0, 0, 0, null, null);
+ super(DUMMY_PATH, 0, 0, 0, 0, null, Collections.emptyList());
this.split = split;
this.tableFormatType = TableFormatType.PAIMON;
@@ -53,7 +54,8 @@ public class PaimonSplit extends FileSplit {
private PaimonSplit(LocationPath file, long start, long length, long
fileLength, long modificationTime,
String[] hosts, List<String> partitionList) {
- super(file, start, length, fileLength, modificationTime, hosts,
partitionList);
+ super(file, start, length, fileLength, modificationTime, hosts,
+ partitionList == null ? Collections.emptyList() :
partitionList);
this.tableFormatType = TableFormatType.PAIMON;
this.selfSplitWeight = length;
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonUtilTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonUtilTest.java
index 046b38e311b..cbddcba24ac 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonUtilTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/PaimonUtilTest.java
@@ -21,6 +21,7 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.Type;
import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.table.Table;
import org.apache.paimon.types.CharType;
@@ -32,6 +33,7 @@ import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -76,4 +78,34 @@ public class PaimonUtilTest {
Assert.assertFalse(partitionInfoMap.containsKey("dt"));
Assert.assertEquals("2026-05-26", partitionInfoMap.get("Dt"));
}
+
+ @Test
+ public void testGetPartitionInfoMapSupportsFloatingPointPartitions() {
+ DataField floatPartition = DataTypes.FIELD(0, "float_partition",
DataTypes.FLOAT());
+ DataField doublePartition = DataTypes.FIELD(1, "double_partition",
DataTypes.DOUBLE());
+ Table table = Mockito.mock(Table.class);
+ Mockito.when(table.name()).thenReturn("mock_table");
+
Mockito.when(table.partitionKeys()).thenReturn(Arrays.asList("float_partition",
"double_partition"));
+ Mockito.when(table.rowType()).thenReturn(DataTypes.ROW(floatPartition,
doublePartition));
+
+ float floatValue = Math.nextUp(0.1F);
+ double doubleValue = Math.nextUp(0.1D);
+ BinaryRow partitionValues = new BinaryRow(2);
+ BinaryRowWriter writer = new BinaryRowWriter(partitionValues);
+ writer.writeFloat(0, floatValue);
+ writer.writeDouble(1, doubleValue);
+ writer.complete();
+
+ Map<String, String> partitionInfoMap = PaimonUtil.getPartitionInfoMap(
+ table, partitionValues, "UTC");
+
+ String serializedFloat = partitionInfoMap.get("float_partition");
+ String serializedDouble = partitionInfoMap.get("double_partition");
+ Assert.assertEquals(Float.toString(floatValue), serializedFloat);
+ Assert.assertEquals(Double.toString(doubleValue), serializedDouble);
+ Assert.assertEquals(Float.floatToIntBits(floatValue),
+ Float.floatToIntBits(Float.parseFloat(serializedFloat)));
+ Assert.assertEquals(Double.doubleToLongBits(doubleValue),
+ Double.doubleToLongBits(Double.parseDouble(serializedDouble)));
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
index 45d8c8909a4..d2d5a8467dc 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/paimon/source/PaimonScanNodeTest.java
@@ -29,10 +29,12 @@ import
org.apache.doris.datasource.paimon.PaimonFileExternalCatalog;
import org.apache.doris.planner.PlanNodeId;
import org.apache.doris.planner.ScanContext;
import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.thrift.TFileRangeDesc;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.stats.SimpleStats;
+import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.RawFile;
import org.junit.Assert;
@@ -66,7 +68,11 @@ public class PaimonScanNodeTest {
TupleDescriptor desc = new TupleDescriptor(new TupleId(3));
PaimonScanNode paimonScanNode = new PaimonScanNode(new PlanNodeId(1),
desc, false, sv, ScanContext.EMPTY);
- paimonScanNode.setSource(new PaimonSource());
+ PaimonSource source = Mockito.spy(new PaimonSource());
+ Table paimonTable = Mockito.mock(Table.class);
+ Mockito.doReturn(paimonTable).when(source).getPaimonTable();
+
Mockito.when(paimonTable.partitionKeys()).thenReturn(Collections.emptyList());
+ paimonScanNode.setSource(source);
DataFileMeta dfm1 = DataFileMeta.forAppend("f1.parquet", 64 * 1024 *
1024, 1, SimpleStats.EMPTY_STATS, 1, 1, 1,
Collections.emptyList(), null, null, null, null);
@@ -473,6 +479,27 @@ public class PaimonScanNodeTest {
Assert.assertEquals(-1, PaimonScanNode.getFieldIndex(fieldNames,
"missing_col"));
}
+ @Test
+ public void testSetPartitionValuesBuildsAlignedMetadata() {
+ PaimonScanNode node = new PaimonScanNode(new PlanNodeId(0), new
TupleDescriptor(new TupleId(0)),
+ false, sv, ScanContext.EMPTY);
+ PaimonSource source = Mockito.mock(PaimonSource.class);
+ Table paimonTable = Mockito.mock(Table.class);
+ Mockito.when(source.getPaimonTable()).thenReturn(paimonTable);
+
Mockito.when(paimonTable.partitionKeys()).thenReturn(Arrays.asList("Region",
"Dt"));
+ node.setSource(source);
+
+ Map<String, String> partitionValues = new HashMap<>();
+ partitionValues.put("Dt", null);
+ partitionValues.put("Region", "cn");
+ TFileRangeDesc rangeDesc = new TFileRangeDesc();
+ node.setPartitionValues(rangeDesc, partitionValues);
+
+ Assert.assertEquals(Arrays.asList("Region", "Dt"),
rangeDesc.getColumnsFromPathKeys());
+ Assert.assertEquals(Arrays.asList("cn", ""),
rangeDesc.getColumnsFromPath());
+ Assert.assertEquals(Arrays.asList(false, true),
rangeDesc.getColumnsFromPathIsNull());
+ }
+
private void mockJniReader(PaimonScanNode spyNode) {
Mockito.doReturn(false).when(spyNode).supportNativeReader(ArgumentMatchers.any(Optional.class));
}
diff --git
a/regression-test/data/external_table_p0/paimon/test_paimon_runtime_filter_partition_pruning.out
b/regression-test/data/external_table_p0/paimon/test_paimon_runtime_filter_partition_pruning.out
index 738d40fabe0..7c4e120d506 100644
---
a/regression-test/data/external_table_p0/paimon/test_paimon_runtime_filter_partition_pruning.out
+++
b/regression-test/data/external_table_p0/paimon/test_paimon_runtime_filter_partition_pruning.out
@@ -32,6 +32,9 @@
-- !runtime_filter_partition_pruning_string_in_null --
2
+-- !static_partition_pruning_string_in --
+6
+
-- !runtime_filter_partition_pruning_date1 --
1
@@ -128,6 +131,9 @@
-- !runtime_filter_partition_pruning_string_in_null --
2
+-- !static_partition_pruning_string_in --
+6
+
-- !runtime_filter_partition_pruning_date1 --
1
@@ -190,4 +196,3 @@
-- !null_partition_4 --
1 \N 100.0
-
diff --git
a/regression-test/suites/external_table_p0/paimon/test_paimon_runtime_filter_partition_pruning.groovy
b/regression-test/suites/external_table_p0/paimon/test_paimon_runtime_filter_partition_pruning.groovy
index 325e9f242d0..e8a3634a2e5 100644
---
a/regression-test/suites/external_table_p0/paimon/test_paimon_runtime_filter_partition_pruning.groovy
+++
b/regression-test/suites/external_table_p0/paimon/test_paimon_runtime_filter_partition_pruning.groovy
@@ -131,6 +131,10 @@ suite("test_paimon_runtime_filter_partition_pruning",
"p0,external,doris,externa
(select partition_key from string_partitioned
order by id desc limit 2);
"""
+ qt_static_partition_pruning_string_in """
+ select count(*) from string_partitioned
+ where partition_key in ('North America', 'Europe', 'Asia');
+ """
qt_runtime_filter_partition_pruning_date1 """
select count(*) from date_partitioned where partition_key =
(select partition_key from date_partitioned
@@ -267,5 +271,3 @@ suite("test_paimon_runtime_filter_partition_pruning",
"p0,external,doris,externa
}
}
}
-
-
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
