This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 7cb4296ac27 branch-4.1: [fix](paimon) preserve partition metadata
across splits (#65581)
7cb4296ac27 is described below
commit 7cb4296ac278e19e12f9088750bd891077ec7e70
Author: Socrates <[email protected]>
AuthorDate: Wed Jul 15 12:14:18 2026 +0800
branch-4.1: [fix](paimon) preserve partition metadata across splits (#65581)
### 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.
- 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.
### Validation
- `clang-format` 16
- `git diff --check`
- FE checkstyle for touched Java files
- Full FE/BE tests will run in CI; local FE UT compilation is blocked by
missing generated Thrift Java classes in this worktree.
---
be/src/exec/scan/file_scanner.cpp | 79 ++++++-------------
be/src/format/generic_reader.h | 3 +-
be/src/format/orc/vorc_reader.cpp | 28 +++----
be/src/format/orc/vorc_reader.h | 4 +-
be/src/format/parquet/vparquet_group_reader.cpp | 24 ++----
be/src/format/parquet/vparquet_group_reader.h | 3 +-
be/src/format/parquet/vparquet_reader.cpp | 4 +-
be/src/format/parquet/vparquet_reader.h | 3 +-
be/src/format/table/partition_column_filler.h | 79 +++++++++++++++++++
be/src/format/table/table_format_reader.h | 6 +-
.../format/table/partition_column_filler_test.cpp | 88 ++++++++++++++++++++++
.../apache/doris/datasource/paimon/PaimonUtil.java | 10 +++
.../datasource/paimon/source/PaimonScanNode.java | 73 ++++++++++++------
.../datasource/paimon/source/PaimonSplit.java | 6 +-
.../doris/datasource/paimon/PaimonUtilTest.java | 34 +++++++++
.../paimon/source/PaimonScanNodeTest.java | 33 +++++++-
...est_paimon_runtime_filter_partition_pruning.out | 7 +-
..._paimon_runtime_filter_partition_pruning.groovy | 8 +-
18 files changed, 366 insertions(+), 126 deletions(-)
diff --git a/be/src/exec/scan/file_scanner.cpp
b/be/src/exec/scan/file_scanner.cpp
index e6cd425bd58..3c7a52a2893 100644
--- a/be/src/exec/scan/file_scanner.cpp
+++ b/be/src/exec/scan/file_scanner.cpp
@@ -75,6 +75,7 @@
#include "format/table/paimon_jni_reader.h"
#include "format/table/paimon_predicate_converter.h"
#include "format/table/paimon_reader.h"
+#include "format/table/partition_column_filler.h"
#include "format/table/remote_doris_reader.h"
#include "format/table/transactional_hive_reader.h"
#include "format/table/trino_connector_jni_reader.h"
@@ -357,33 +358,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);
}
@@ -395,20 +375,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;
}
@@ -767,22 +736,10 @@ Status FileScanner::_fill_columns_from_path(size_t rows) {
auto column_guard =
_src_block_ptr->mutate_column_scoped(_src_block_name_to_idx[kv.first]);
IColumn* col_ptr = column_guard.mutable_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();
}
@@ -1630,7 +1587,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));
@@ -1766,6 +1724,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());
@@ -1773,6 +1734,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/src/format/generic_reader.h b/be/src/format/generic_reader.h
index e81358ed36d..266244574ec 100644
--- a/be/src/format/generic_reader.h
+++ b/be/src/format/generic_reader.h
@@ -78,7 +78,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/format/orc/vorc_reader.cpp
b/be/src/format/orc/vorc_reader.cpp
index f62c4a43715..8b42d2cceb3 100644
--- a/be/src/format/orc/vorc_reader.cpp
+++ b/be/src/format/orc/vorc_reader.cpp
@@ -82,6 +82,7 @@
#include "exprs/vruntimefilter_wrapper.h"
#include "format/orc/orc_file_reader.h"
#include "format/table/iceberg_reader.h"
+#include "format/table/partition_column_filler.h"
#include "format/table/transactional_hive_common.h"
#include "io/fs/buffered_reader.h"
#include "io/fs/file_reader.h"
@@ -1194,8 +1195,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;
@@ -1531,23 +1534,12 @@ Status OrcReader::_fill_partition_columns(
auto column_guard =
block->mutate_column_scoped((*_col_name_to_block_idx)[kv.first]);
auto& col_ptr = column_guard.mutable_column();
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/format/orc/vorc_reader.h b/be/src/format/orc/vorc_reader.h
index cfbd7abb8cb..7305e5637ff 100644
--- a/be/src/format/orc/vorc_reader.h
+++ b/be/src/format/orc/vorc_reader.h
@@ -106,6 +106,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;
@@ -177,7 +178,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/format/parquet/vparquet_group_reader.cpp
b/be/src/format/parquet/vparquet_group_reader.cpp
index 7867841593d..7c534a909cb 100644
--- a/be/src/format/parquet/vparquet_group_reader.cpp
+++ b/be/src/format/parquet/vparquet_group_reader.cpp
@@ -61,6 +61,7 @@
#include "format/parquet/schema_desc.h"
#include "format/parquet/vparquet_column_reader.h"
#include "format/table/iceberg_reader.h"
+#include "format/table/partition_column_filler.h"
#include "runtime/descriptors.h"
#include "runtime/runtime_state.h"
#include "runtime/thread_context.h"
@@ -835,23 +836,12 @@ Status RowGroupReader::_fill_partition_columns(
auto column_guard = block->mutate_column_scoped(block_pos);
auto* col_ptr = column_guard.mutable_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/format/parquet/vparquet_group_reader.h
b/be/src/format/parquet/vparquet_group_reader.h
index c103a8d8102..4d862954e85 100644
--- a/be/src/format/parquet/vparquet_group_reader.h
+++ b/be/src/format/parquet/vparquet_group_reader.h
@@ -88,9 +88,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/format/parquet/vparquet_reader.cpp
b/be/src/format/parquet/vparquet_reader.cpp
index 3ce33fb1eac..a1e376ce55e 100644
--- a/be/src/format/parquet/vparquet_reader.cpp
+++ b/be/src/format/parquet/vparquet_reader.cpp
@@ -502,8 +502,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/format/parquet/vparquet_reader.h
b/be/src/format/parquet/vparquet_reader.h
index 868ff8927d8..5284447a0da 100644
--- a/be/src/format/parquet/vparquet_reader.h
+++ b/be/src/format/parquet/vparquet_reader.h
@@ -172,7 +172,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/format/table/partition_column_filler.h
b/be/src/format/table/partition_column_filler.h
new file mode 100644
index 00000000000..0f38a67d3ab
--- /dev/null
+++ b/be/src/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 "core/assert_cast.h"
+#include "core/column/column_nullable.h"
+#include "core/data_type_serde/data_type_serde.h"
+#include "runtime/descriptors.h"
+#include "util/slice.h"
+
+namespace doris {
+
+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
diff --git a/be/src/format/table/table_format_reader.h
b/be/src/format/table/table_format_reader.h
index f5f0e6e953e..8ed4ad70c11 100644
--- a/be/src/format/table/table_format_reader.h
+++ b/be/src/format/table/table_format_reader.h
@@ -94,8 +94,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/test/format/table/partition_column_filler_test.cpp
b/be/test/format/table/partition_column_filler_test.cpp
new file mode 100644
index 00000000000..f7cd9b80f0a
--- /dev/null
+++ b/be/test/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 "format/table/partition_column_filler.h"
+
+#include <gtest/gtest.h>
+
+#include <memory>
+
+#include "core/assert_cast.h"
+#include "core/column/column_nullable.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "core/data_type/data_type_string.h"
+
+namespace doris {
+
+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
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 ae26cf7aec4..2a738a0ac8d 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
@@ -577,6 +577,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 13f38790933..f465e3c9534 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
@@ -52,6 +52,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;
@@ -70,6 +71,7 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
@@ -233,6 +235,52 @@ public class PaimonScanNode extends FileQueryScanNode {
}
}
+ private List<String> getOrderedPathPartitionKeys() {
+ if (source == null) {
+ return Collections.emptyList();
+ }
+ ExternalTable externalTable = source.getExternalTable();
+ if (externalTable instanceof PaimonSysExternalTable
+ && !((PaimonSysExternalTable) externalTable).isDataTable()) {
+ return Collections.emptyList();
+ }
+ return source.getPaimonTable().partitionKeys().stream()
+ .map(key -> key.toLowerCase(Locale.ROOT))
+ .collect(Collectors.toList());
+ }
+
+ @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");
+
+ Map<String, String> normalizedPartitionValues = new HashMap<>();
+ for (Map.Entry<String, String> entry : partitionValues.entrySet()) {
+
normalizedPartitionValues.put(entry.getKey().toLowerCase(Locale.ROOT),
entry.getValue());
+ }
+
+ List<String> fromPathValues = new
ArrayList<>(orderedPartitionKeys.size());
+ List<Boolean> fromPathIsNull = new
ArrayList<>(orderedPartitionKeys.size());
+ for (String partitionKey : orderedPartitionKeys) {
+
Preconditions.checkState(normalizedPartitionValues.containsKey(partitionKey),
+ "Missing partition value for Paimon partition key: %s",
partitionKey);
+ String partitionValue =
normalizedPartitionValues.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());
@@ -290,20 +338,7 @@ public class PaimonScanNode extends FileQueryScanNode {
tableFormatFileDesc.setTableLevelRowCount(-1);
}
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);
}
@@ -371,6 +406,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;
@@ -380,9 +416,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());
@@ -605,10 +639,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 f9d3e2e4ae9..4a8808517b2 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
@@ -27,6 +27,7 @@ import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.DeletionFile;
import org.apache.paimon.table.source.Split;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -47,7 +48,7 @@ public class PaimonSplit extends FileSplit {
* Handles both DataSplit (regular data tables) and other Split types
(system tables).
*/
public PaimonSplit(Split paimonSplit) {
- super(DUMMY_PATH, 0, 0, 0, 0, null, null);
+ super(DUMMY_PATH, 0, 0, 0, 0, null, Collections.emptyList());
this.paimonSplit = paimonSplit;
this.tableFormatType = TableFormatType.PAIMON;
@@ -65,7 +66,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 1d519c255f8..ac5ceab8049 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
@@ -22,7 +22,10 @@ import org.apache.doris.thrift.TPrimitiveType;
import org.apache.doris.thrift.schema.external.TFieldPtr;
import org.apache.doris.thrift.schema.external.TSchema;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.BinaryRowWriter;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.Table;
import org.apache.paimon.types.CharType;
import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
@@ -34,6 +37,7 @@ import org.mockito.Mockito;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
public class PaimonUtilTest {
private static final String TABLE_READ_SEQUENCE_NUMBER_ENABLED =
"table-read.sequence-number.enabled";
@@ -49,6 +53,36 @@ public class PaimonUtilTest {
Assert.assertEquals(14, type2.getLength());
}
+ @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)));
+ }
+
@Test
public void testBinlogHistorySchemaWithSequenceNumber() {
PaimonSysExternalTable binlogTable =
Mockito.mock(PaimonSysExternalTable.class);
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 f0e8a91d360..c0da0fadc46 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
@@ -39,6 +39,7 @@ import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.DataFileMeta;
import org.apache.paimon.manifest.FileSource;
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;
@@ -51,6 +52,7 @@ import org.mockito.junit.MockitoJUnitRunner;
import java.lang.reflect.Method;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -71,7 +73,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", 64L * 1024 *
1024, 1L, SimpleStats.EMPTY_STATS,
1L, 1L, 1L, Collections.<String>emptyList(), null,
FileSource.APPEND,
@@ -439,7 +445,6 @@ public class PaimonScanNodeTest {
Mockito.when(sv.isForceJniScanner()).thenReturn(false);
Mockito.when(sv.getIgnoreSplitType()).thenReturn("NONE");
-
Mockito.when(sv.isEnableRuntimeFilterPartitionPrune()).thenReturn(false);
Mockito.when(sv.getMaxSplitSize()).thenReturn(maxSplitSize);
Assert.assertTrue(spyPaimonScanNode.shouldForceJniForSystemTable());
@@ -518,7 +523,10 @@ public class PaimonScanNodeTest {
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.getTableLocation()).thenReturn("file:///warehouse");
+ Mockito.when(source.getPaimonTable()).thenReturn(paimonTable);
+
Mockito.when(paimonTable.partitionKeys()).thenReturn(Collections.emptyList());
node.setSource(source);
Map<String, String> backendOptions = new HashMap<>();
@@ -539,6 +547,27 @@ public class PaimonScanNodeTest {
Assert.assertFalse(rangeDesc.getTableFormatParams().getPaimonParams().isSetPaimonOptions());
}
+ @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 f7a666d2c83..5e289e8d919 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
@@ -102,6 +102,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
@@ -227,6 +231,7 @@ suite("test_paimon_runtime_filter_partition_pruning",
"p0,external,doris,externa
try {
sql """ set time_zone = 'Asia/Shanghai'; """
+ sql """ set enable_file_scanner_v2 = false; """
sql """ set enable_runtime_filter_partition_prune = false; """
test_runtime_filter_partition_pruning()
sql """ set enable_runtime_filter_partition_prune = true; """
@@ -234,9 +239,8 @@ suite("test_paimon_runtime_filter_partition_pruning",
"p0,external,doris,externa
} finally {
sql """ unset variable time_zone; """
+ sql """ unset variable enable_file_scanner_v2; """
sql """ set enable_runtime_filter_partition_prune = true; """
}
}
}
-
-
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
