This is an automated email from the ASF dual-hosted git repository.
Gabriel39 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 5d46ecfd7e3 [fix](external) Backport SCHEMATA, ORC and JSON/WAL fixes
to 4.1 (#70319)
5d46ecfd7e3 is described below
commit 5d46ecfd7e376d1422cad8b9addeda1a089f326d
Author: Gabriel <[email protected]>
AuthorDate: Mon Nov 9 16:56:07 2026 +0800
[fix](external) Backport SCHEMATA, ORC and JSON/WAL fixes to 4.1 (#70319)
### What problem does this PR solve?
Backport three master fixes to `branch-4.1`: avoid unnecessary
database-object loading for SCHEMATA scans, prevent ORC condition-cache
seek crashes or missing rows, and preserve JSON/WAL row counts when no
file columns are projected.
The commits retain their original authors and are applied in master
merge order:
1. https://github.com/apache/doris/pull/70010 — Avoid database-object
enumeration when SCHEMATA only needs names.
2. https://github.com/apache/doris/pull/70079 — Advance safely past
SARG-filtered ORC stripe suffixes during condition-cache seeks.
3. https://github.com/apache/doris/pull/70154 — Track logical JSON
records independently of projected columns and capture WAL row counts
before column ownership is transferred.
Branch adaptations:
- Keep Thrift field 7 reserved and add the optional `include_db_ids`
flag at upstream field 8 with its default `true`. Preserve existing
privilege checks, pattern matching and schema-name formatting for both
enumeration paths.
- Use the upstream fixed ORC dependency revision
`5962bb071d6345cc29c5353d64d85a203525e4d9`, which descends from the
branch's existing revision.
- Resolve the JSON reader conflicts by retaining 4.1's error handling
and adding logical row accounting without importing unrelated
missing-field/load-rejection changes. Remove the rooted-array pre-count
that consumes the ondemand cursor before JSON-path materialization;
retain the upstream rooted-array coverage.
### Validation
- The branch's FE production and test sources compiled with the existing
Java 8 target settings. All **7 targeted FE tests passed**, including
the existing `FrontendServiceImplTest#testGetDBNames`, the
100,000-namespace case, filtering and legacy-request compatibility. FE
Checkstyle passed.
- A fresh ASAN build of the exact pinned ORC dependency passed **56
focused tests**, including **32 filtered-stripe-suffix boundary cases**.
- All **7 affected C++ implementation/test/generator translation units
compiled**. Clang-format 16 passed for all **8 affected C++
source/header files**.
- The ORC fixture generator reproduced the committed fixture byte for
byte. All **3 added Groovy suites compiled**.
- Full Doris BE unit-test execution and external SQL regression
execution remain for CI; they were not run locally for this backport.
### Release note
Improve SCHEMATA enumeration for large catalogs. Fix potential crashes
or missing rows during ORC condition-cache seeks and incorrect JSON/WAL
row counts with empty file projections.
### Check List (For Author)
- Test
- [x] Unit tests and focused dependency tests
- [x] Regression tests included and compiled; full SQL execution pending
CI
- Behavior changed:
- [x] Yes. Apply the upstream fixes while preserving branch-specific
validation and protocol compatibility.
- Does this need documentation?
- [x] No.
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Add branch pick label
### ARM64 link dependency follow-up
The ORC revision enables libdeflate ZLIB decoding on every architecture.
Remove the old x86-only guard around `add_thirdparty(deflate)` so ARM64
BE links also receive the required library, matching master. An adjacent
comment records this dependency invariant.
Validation: probes include each branch's actual
`be/cmake/thirdparty.cmake` and select `aarch64`, `arm64`, `x86` and
`x86_64` dependency configurations. Before the fix, both ARM
configurations fail to link with missing libdeflate symbols while both
x86 controls pass. After the fix, all eight repository/configuration
combinations link and pass a DEFLATE compression/decompression round
trip. These probes run natively on x86_64 and validate CMake dependency
selection; a native ARM64/full BE build remains for CI. Clang-format 16
passed for the PR-affected C++ files.
---
be/cmake/thirdparty.cmake | 6 +-
be/src/format_v2/json/json_reader.cpp | 55 ++++--
be/src/format_v2/json/json_reader.h | 3 +
be/src/format_v2/wal/wal_reader.cpp | 5 +-
.../information_schema/schema_schemata_scanner.cpp | 2 +
be/test/format_v2/json/json_reader_test.cpp | 162 ++++++++++++++++++
be/test/format_v2/orc/orc_reader_test.cpp | 66 ++++++++
be/test/format_v2/wal/wal_reader_test.cpp | 120 +++++++++++++
contrib/apache-orc | 2 +-
.../apache/doris/service/FrontendServiceImpl.java | 54 ++++--
.../service/FrontendServiceGetDbNamesTest.java | 186 +++++++++++++++++++++
gensrc/thrift/FrontendService.thrift | 2 +
.../test_data/GenerateOrcConditionCacheSeek.cpp | 51 ++++++
.../tvf/test_data/condition_cache_seek.orc | Bin 0 -> 1458 bytes
.../test_group_commit_wal_row_count.out | 17 ++
.../hive/test_hive_json_partition_only.groovy | 85 ++++++++++
.../tvf/test_orc_condition_cache_seek.groovy | 110 ++++++++++++
.../test_group_commit_wal_row_count.groovy | 85 ++++++++++
18 files changed, 976 insertions(+), 35 deletions(-)
diff --git a/be/cmake/thirdparty.cmake b/be/cmake/thirdparty.cmake
index b67c8e9d0aa..5471fa30979 100644
--- a/be/cmake/thirdparty.cmake
+++ b/be/cmake/thirdparty.cmake
@@ -186,10 +186,8 @@ if (OS_MACOSX)
add_thirdparty(intl)
endif()
-# Only used on x86 or x86_64
-if ("${CMAKE_BUILD_TARGET_ARCH}" STREQUAL "x86" OR
"${CMAKE_BUILD_TARGET_ARCH}" STREQUAL "x86_64")
- add_thirdparty(deflate)
-endif()
+# ORC's ZLIB decoder requires libdeflate on every architecture, including
ARM64.
+add_thirdparty(deflate)
add_thirdparty(icuuc LIB64)
add_thirdparty(icui18n LIB64)
diff --git a/be/src/format_v2/json/json_reader.cpp
b/be/src/format_v2/json/json_reader.cpp
index 78601f90584..e2a963093fa 100644
--- a/be/src/format_v2/json/json_reader.cpp
+++ b/be/src/format_v2/json/json_reader.cpp
@@ -306,13 +306,13 @@ Status JsonReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
return Status::InternalError("JSON v2 reader is not open");
}
- const auto batch_size = _runtime_state->batch_size();
+ const auto batch_size = cast_set<size_t>(_runtime_state->batch_size());
const auto max_block_bytes = _runtime_state->preferred_block_size_bytes();
*rows = 0;
*eof = false;
+ _logical_rows = 0;
- while (file_block->rows() < batch_size && !_reader_eof &&
- file_block->bytes() < max_block_bytes) {
+ while (_logical_rows < batch_size && !_reader_eof && file_block->bytes() <
max_block_bytes) {
if (_read_json_by_line && _skip_first_line) {
RETURN_IF_ERROR(_line_reader->skip_split_prefix(
_reader_range.start_offset, _line_delimiter, &_reader_eof,
_io_ctx.get()));
@@ -320,7 +320,7 @@ Status JsonReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
continue;
}
- const size_t original_rows = file_block->rows();
+ const size_t original_rows = _logical_rows;
size_t size = 0;
bool is_empty_row = false;
Status st = Status::OK();
@@ -349,12 +349,13 @@ Status JsonReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
}
// An ignored or empty JSON object can produce no row. Avoid spinning
forever on a document
// that was consumed but produced no materialized value.
- if (!is_empty_row && file_block->rows() == original_rows) {
+ if (!is_empty_row && _logical_rows == original_rows) {
break;
}
}
- *rows = file_block->rows();
+ *rows = _logical_rows;
+ DCHECK(file_block->columns() == 0 || file_block->rows() == *rows);
_record_scan_rows(cast_set<int64_t>(*rows));
{
SCOPED_TIMER(_filter_time);
@@ -671,10 +672,8 @@ Status JsonReader::_extract_json_value(size_t size, bool*
eof, bool* is_empty_ro
return Status::DataQualityError(
"JSON data is not an array-object, `strip_outer_array` must be
FALSE.");
}
- if (!_parsed_jsonpaths.empty() && _strip_outer_array &&
- _json_value.count_elements().value() == 0) {
- *is_empty_row = true;
- }
+ // A rooted ondemand array has one cursor. Counting it here would consume
records
+ // before jsonpaths materialization; the append path detects an empty
array itself.
return Status::OK();
}
@@ -725,7 +724,7 @@ Status JsonReader::_append_flat_array_jsonpath_rows(Block*
block, bool* is_empty
DORIS_CHECK(block != nullptr);
DORIS_CHECK(is_empty_row != nullptr);
DORIS_CHECK(eof != nullptr);
- const size_t original_rows = block->rows();
+ const size_t original_rows = _logical_rows;
bool valid = true;
_array = _json_value.get_array();
_array_iter = _array.begin();
@@ -753,7 +752,7 @@ Status JsonReader::_append_flat_array_jsonpath_rows(Block*
block, bool* is_empty
RETURN_IF_ERROR(_write_columns_by_jsonpath(&object_value, block,
&valid));
++_array_iter;
}
- *is_empty_row = block->rows() == original_rows;
+ *is_empty_row = _logical_rows == original_rows;
return Status::OK();
}
@@ -777,7 +776,7 @@ Status
JsonReader::_set_column_values_from_object(simdjson::ondemand::object* ob
DORIS_CHECK(block != nullptr);
DORIS_CHECK(valid != nullptr);
std::vector<bool> seen_columns(block->columns(), false);
- const size_t cur_row_count = block->rows();
+ const size_t cur_row_count = _logical_rows;
bool has_valid_value = false;
size_t key_index = 0;
@@ -821,21 +820,46 @@ Status
JsonReader::_set_column_values_from_object(simdjson::ondemand::object* ob
}
}
*valid = true;
+ ++_logical_rows;
if (!has_valid_value) {
return Status::OK();
}
return Status::OK();
}
+Status JsonReader::_has_matching_jsonpath(simdjson::ondemand::object*
object_value,
+ bool* matches) const {
+ *matches = false;
+ for (const auto& path : _parsed_jsonpaths) {
+ // A root path selects the current document without a field lookup.
+ if (JsonFunctions::is_root_path(path)) {
+ *matches = true;
+ continue;
+ }
+ simdjson::ondemand::value value;
+ const auto status = JsonFunctions::extract_from_object(*object_value,
path, &value);
+ if (!status.ok() && !status.is<ErrorCode::NOT_FOUND>()) {
+ return status;
+ }
+ *matches = *matches || status.ok();
+ }
+ return Status::OK();
+}
+
Status JsonReader::_write_columns_by_jsonpath(simdjson::ondemand::object*
object_value,
Block* block, bool* valid) {
DORIS_CHECK(object_value != nullptr);
DORIS_CHECK(block != nullptr);
DORIS_CHECK(valid != nullptr);
bool has_valid_value = false;
- const size_t cur_row_count = block->rows();
+ const size_t cur_row_count = _logical_rows;
_cached_string_values.clear();
+ // Even without projected fields, jsonpaths still decides whether the
object contributes a row.
+ if (_requested_columns.empty()) {
+ RETURN_IF_ERROR(_has_matching_jsonpath(object_value,
&has_valid_value));
+ }
+
for (size_t i = 0; i < _requested_columns.size(); ++i) {
const auto& requested = _requested_columns[i];
auto* column_ptr =
block->get_by_position(i).column->assert_mutable().get();
@@ -889,6 +913,7 @@ Status
JsonReader::_write_columns_by_jsonpath(simdjson::ondemand::object* object
return Status::OK();
}
*valid = true;
+ ++_logical_rows;
return Status::OK();
}
@@ -1086,6 +1111,7 @@ Status JsonReader::_append_null_for_malformed_json(Block*
block) {
assert_cast<ColumnNullable*>(column.get())->insert_default();
column_with_type.column = std::move(column);
}
+ ++_logical_rows;
return Status::OK();
}
@@ -1097,6 +1123,7 @@ Status JsonReader::_handle_json_error(const Status&
status, Block* block, size_t
// block to the row count before this document before either surfacing the
error or appending
// the ignore-malformed null row.
_truncate_block_to_rows(block, original_rows);
+ _logical_rows = original_rows;
if (_openx_json_ignore_malformed &&
status.is<ErrorCode::DATA_QUALITY_ERROR>()) {
RETURN_IF_ERROR(_append_null_for_malformed_json(block));
*is_empty_row = false;
diff --git a/be/src/format_v2/json/json_reader.h
b/be/src/format_v2/json/json_reader.h
index f5e6614b2da..075d5aca534 100644
--- a/be/src/format_v2/json/json_reader.h
+++ b/be/src/format_v2/json/json_reader.h
@@ -147,6 +147,7 @@ private:
Status _append_nested_jsonpath_row(Block* block, bool* is_empty_row, bool*
eof);
Status _set_column_values_from_object(simdjson::ondemand::object*
object_value, Block* block,
bool* valid);
+ Status _has_matching_jsonpath(simdjson::ondemand::object* object_value,
bool* matches) const;
Status _write_columns_by_jsonpath(simdjson::ondemand::object*
object_value, Block* block,
bool* valid);
template <bool use_string_cache>
@@ -190,6 +191,8 @@ private:
std::unique_ptr<Decompressor> _decompressor;
std::unique_ptr<NewPlainTextLineReader> _line_reader;
int64_t _current_offset = 0;
+ // Partition-only scans have no file columns, so Block::rows() cannot
count their records.
+ size_t _logical_rows = 0;
bool _reader_eof = false;
bool _skip_first_line = false;
bool _single_document_read = false;
diff --git a/be/src/format_v2/wal/wal_reader.cpp
b/be/src/format_v2/wal/wal_reader.cpp
index 55e88796c98..5f57a422ff8 100644
--- a/be/src/format_v2/wal/wal_reader.cpp
+++ b/be/src/format_v2/wal/wal_reader.cpp
@@ -195,8 +195,11 @@ Status WalReader::get_block(Block* file_block, size_t*
rows, bool* eof) {
return Status::Corruption("WAL block has {} columns but header
declares {}",
source_block.columns(), _column_ids.size());
}
+ // Empty projections still contribute every WAL row. Capture the count
before materialization
+ // moves source columns into the output block and can leave the source
with no owned columns.
+ const size_t source_rows = source_block.rows();
RETURN_IF_ERROR(_materialize_requested_columns(&source_block, file_block));
- *rows = file_block->rows();
+ *rows = source_rows;
_record_scan_rows(cast_set<int64_t>(*rows));
RETURN_IF_ERROR(
apply_materialized_reader_filters(_request.get(), _io_ctx.get(),
file_block, rows));
diff --git a/be/src/information_schema/schema_schemata_scanner.cpp
b/be/src/information_schema/schema_schemata_scanner.cpp
index e9d25684780..15e2528a89f 100644
--- a/be/src/information_schema/schema_schemata_scanner.cpp
+++ b/be/src/information_schema/schema_schemata_scanner.cpp
@@ -52,6 +52,8 @@ Status SchemaSchemataScanner::start(RuntimeState* state) {
return Status::InternalError("used before initial.");
}
TGetDbsParams db_params;
+ // SCHEMATA never reads database IDs; avoid creating one database object
per namespace in FE.
+ db_params.__set_include_db_ids(false);
if (nullptr != _param->common_param->wild) {
db_params.__set_pattern(*(_param->common_param->wild));
}
diff --git a/be/test/format_v2/json/json_reader_test.cpp
b/be/test/format_v2/json/json_reader_test.cpp
index 1d94e244f64..f25fd3d8b05 100644
--- a/be/test/format_v2/json/json_reader_test.cpp
+++ b/be/test/format_v2/json/json_reader_test.cpp
@@ -43,6 +43,7 @@
#include "exprs/vexpr_context.h"
#include "format/json/new_json_reader.h"
#include "format_v2/column_data.h"
+#include "format_v2/table_reader.h"
#include "io/io_common.h"
#include "runtime/descriptors.h"
#include "runtime/runtime_profile.h"
@@ -802,4 +803,165 @@ TEST(JsonReaderTest,
AppliesDeleteAndNormalConjunctsWithPredicateFilterAccountin
EXPECT_EQ(io_ctx->predicate_filtered_rows, 1);
}
+// Partition-only scans have no file-local output columns, but still need one
logical row per
+// accepted JSON record so TableReader can materialize the partition constants.
+TEST(JsonReaderTest, EmptyFileProjectionPreservesRows) {
+ ObjectPool pool;
+ const auto slots = build_slots(&pool);
+ for (const auto& source_slots : {slots, std::vector<SlotDescriptor*> {}}) {
+ auto result = read_once("empty_projection.jsonl",
"{\"id\":1}\n\n{}\n{\"id\":3}\n",
+ json_scan_params(), source_slots, {}, true,
true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.block.columns(), 0);
+ EXPECT_EQ(result.rows, 3);
+ EXPECT_FALSE(result.eof);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_EQ(result.second_rows, 0);
+ EXPECT_TRUE(result.second_eof);
+ }
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesOuterArrayRows) {
+ auto result = read_once("empty_projection_array.json",
R"([{"id":1},{},{"id":3}])",
+ json_scan_params(false, true), {}, {}, true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 3);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesJsonRootRows) {
+ auto result = read_once("empty_projection_root.jsonl",
+
"{\"payload\":{\"id\":1}}\n{\"payload\":{\"id\":2}}\n",
+ json_scan_params(true, false, "", "$.payload"),
{}, {}, true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionHonorsJsonPaths) {
+ ObjectPool pool;
+ const auto slots = build_slots(&pool);
+ auto result = read_once("empty_projection_jsonpaths.jsonl",
+
"{\"payload\":{\"id\":1}}\n{\"other\":2}\n{\"payload\":{\"id\":3}}\n",
+ json_scan_params(true, false,
R"(["$.payload.id","$.payload.user"])"),
+ slots, {}, true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesRootJsonPathRows) {
+ auto result = read_once("empty_projection_root_path.jsonl",
"{\"id\":1}\n{}\n",
+ json_scan_params(true, false, R"(["$."])"), {},
{}, true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesRootedArrayJsonPathRows) {
+ auto result = read_once("empty_projection_rooted_array.json",
+
R"({"rows":[{"payload":{"id":1}},{"other":2},{"payload":{"id":3}}]})",
+ json_scan_params(false, true,
R"(["$.payload.id"])", "$.rows"), {}, {},
+ true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesMalformedNullRows) {
+ auto result = read_once("empty_projection_malformed.jsonl",
"not-json\n{\"id\":1}\n",
+ json_scan_params(true, false, "", "", true), {},
{}, true, true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 2);
+ EXPECT_TRUE(result.second_status.ok());
+ EXPECT_TRUE(result.second_eof);
+
+ auto strict = read_once("empty_projection_strict.jsonl", "not-json\n",
json_scan_params(), {},
+ {}, false, true);
+ EXPECT_FALSE(strict.status.ok());
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionPreservesEmptyInput) {
+ for (const std::string content : {"", "\n", "[]\n"}) {
+ auto result = read_once("empty_projection_empty.jsonl", content,
+ json_scan_params(true, true), {}, {}, false,
true);
+ ASSERT_TRUE(result.status.ok()) << result.status;
+ EXPECT_EQ(result.rows, 0);
+ EXPECT_TRUE(result.eof);
+ }
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionHonorsBatchSize) {
+ const auto path = write_json_file("empty_projection_batches.jsonl",
+ "{\"id\":1}\n{\"id\":2}\n{\"id\":3}\n"
+ "{\"id\":4}\n{\"id\":5}\n");
+ auto properties = std::make_shared<io::FileSystemProperties>();
+ properties->system_type = TFileType::FILE_LOCAL;
+ auto desc = file_description(path.string());
+ auto params = json_scan_params();
+ RuntimeProfile profile("json_empty_projection_batches");
+ MockRuntimeState state;
+ state._batch_size = 2;
+ JsonReader reader(properties, desc, std::make_shared<io::IOContext>(),
&profile, ¶ms,
+ file_range(path), {});
+ ASSERT_TRUE(reader.init(&state).ok());
+ ASSERT_TRUE(reader.open(std::make_shared<FileScanRequest>()).ok());
+ for (const size_t expected_rows : {2, 2, 1, 0}) {
+ Block block;
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader.get_block(&block, &rows, &eof).ok());
+ EXPECT_EQ(block.columns(), 0);
+ EXPECT_EQ(rows, expected_rows);
+ EXPECT_EQ(eof, expected_rows == 0);
+ }
+}
+
+TEST(JsonReaderTest, EmptyFileProjectionMaterializesPartitionRows) {
+ const auto path =
+ write_json_file("partition_only.jsonl",
"{\"id\":1}\n{\"id\":2}\n{\"id\":3}\n");
+ auto params = json_scan_params();
+ const std::vector<SlotDescriptor*> file_slots;
+ ColumnDefinition partition;
+ partition.identifier = Field::create_field<TYPE_STRING>("dt");
+ partition.local_id = 0;
+ partition.name = "dt";
+ partition.type = std::make_shared<DataTypeString>();
+ partition.is_partition_key = true;
+ RuntimeProfile profile("json_partition_only");
+ MockRuntimeState state;
+ state._batch_size = 2;
+ TableReader reader;
+ ASSERT_TRUE(reader.init({.projected_columns = {partition},
+ .conjuncts = {},
+ .format = FileFormat::JSON,
+ .scan_params = ¶ms,
+ .io_ctx = std::make_shared<io::IOContext>(),
+ .runtime_state = &state,
+ .scanner_profile = &profile,
+ .file_slot_descs = &file_slots})
+ .ok());
+ SplitReadOptions split;
+ split.current_range = file_range(path);
+ split.current_split_format = FileFormat::JSON;
+ split.partition_values.emplace("dt",
Field::create_field<TYPE_STRING>("p1"));
+ ASSERT_TRUE(reader.prepare_split(split).ok());
+ bool eos = false;
+ for (const size_t expected_rows : {2, 1, 0}) {
+ Block block({{partition.type->create_column(), partition.type,
partition.name}});
+ ASSERT_TRUE(reader.get_block(&block, &eos).ok());
+ ASSERT_EQ(block.rows(), expected_rows);
+ EXPECT_EQ(eos, expected_rows == 0);
+ const auto column =
block.get_by_position(0).column->convert_to_full_column_if_const();
+ for (size_t row = 0; row < expected_rows; ++row) {
+ EXPECT_EQ(column->get_data_at(row).to_string(), "p1");
+ }
+ }
+ EXPECT_TRUE(reader.close().ok());
+}
+
} // namespace doris::format::json
diff --git a/be/test/format_v2/orc/orc_reader_test.cpp
b/be/test/format_v2/orc/orc_reader_test.cpp
index 1817793936e..28564cf88dd 100644
--- a/be/test/format_v2/orc/orc_reader_test.cpp
+++ b/be/test/format_v2/orc/orc_reader_test.cpp
@@ -6313,6 +6313,72 @@ TEST_F(NewOrcReaderTest,
ConditionCacheHitSkipsFalseGranulesBeforeColumnRead) {
EXPECT_EQ(rows, 0);
}
+TEST_F(NewOrcReaderTest, ConditionCacheHitSeeksPastSargFilteredStripeSuffix) {
+ const auto file_path = find_repo_file(
+
"regression-test/data/external_table_p0/tvf/test_data/condition_cache_seek.orc");
+ ASSERT_FALSE(file_path.empty());
+ for (const bool lazy : {false, true}) {
+ for (const bool min_max : {false, true}) {
+ SCOPED_TRACE(fmt::format("lazy={}, min_max={}", lazy, min_max));
+ auto io_ctx = std::make_shared<io::IOContext>();
+ auto reader = create_reader_for_path(file_path.string(), nullptr,
io_ctx);
+ TQueryOptions options;
+ options.__set_enable_orc_lazy_mat(lazy);
+ options.__set_enable_orc_filter_by_min_max(min_max);
+ RuntimeState state {options, TQueryGlobals()};
+ ASSERT_TRUE(reader->init(&state).ok());
+ reader->set_batch_size(ConditionCacheContext::GRANULE_SIZE);
+
+ std::vector<format::ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ auto request = std::make_shared<format::FileScanRequest>();
+ request->predicate_columns = {field_projection(0)};
+ request->non_predicate_columns = {field_projection(1)};
+ request->local_positions = {{format::LocalColumnId(0),
format::LocalIndex(0)},
+ {format::LocalColumnId(1),
format::LocalIndex(1)}};
+ request->conjuncts.push_back(VExprContext::create_shared(
+ std::make_shared<NullableInt32GreaterThanExpr>(0, 5000)));
+ ASSERT_TRUE(reader->open(request).ok());
+
+ auto ctx = std::make_shared<ConditionCacheContext>();
+ ctx->is_hit = true;
+ ctx->filter_result = std::make_shared<std::vector<bool>>(20,
false);
+ (*ctx->filter_result)[0] = true;
+ (*ctx->filter_result)[9] = true;
+ reader->set_condition_cache_context(ctx);
+ int seeks = 0;
+ ScopedDebugPoint debug_point(
+
"OrcReader._skip_condition_cache_false_granules.before_seek_to_row",
+ [&]() { ++seeks; });
+
+ // Granule 9 contains the next stripe's matching row, but starts
in the
+ // current stripe's filtered suffix. Both columns must resume in
stripe 2.
+ std::vector<int32_t> ids;
+ std::vector<int64_t> payloads;
+ bool eof = false;
+ while (!eof) {
+ Block block = build_file_block(schema);
+ size_t rows = 0;
+ ASSERT_TRUE(reader->get_block(&block, &rows, &eof).ok());
+ const auto& id_column = assert_cast<const ColumnInt32&>(
+ assert_cast<const
ColumnNullable&>(*block.get_by_position(0).column)
+ .get_nested_column());
+ const auto& payload_column = assert_cast<const ColumnInt64&>(
+ assert_cast<const
ColumnNullable&>(*block.get_by_position(1).column)
+ .get_nested_column());
+ for (size_t row = 0; row < rows; ++row) {
+ ids.push_back(id_column.get_element(row));
+ payloads.push_back(payload_column.get_element(row));
+ }
+ }
+ EXPECT_EQ(ids, (std::vector<int32_t> {10000, 10001}));
+ EXPECT_EQ(payloads, (std::vector<int64_t> {0, 20000}));
+ EXPECT_GT(seeks, 0);
+ EXPECT_GT(io_ctx->condition_cache_filtered_rows, 0);
+ }
+ }
+}
+
TEST_F(NewOrcReaderTest, ConditionCacheSeekReturnsCleanEofWhenCancelled) {
constexpr int64_t row_count = ConditionCacheContext::GRANULE_SIZE * 2;
const auto file_path = (_test_dir /
"condition_cache_cancelled_seek.orc").string();
diff --git a/be/test/format_v2/wal/wal_reader_test.cpp
b/be/test/format_v2/wal/wal_reader_test.cpp
index 77c14afb12f..65e37bd5f89 100644
--- a/be/test/format_v2/wal/wal_reader_test.cpp
+++ b/be/test/format_v2/wal/wal_reader_test.cpp
@@ -22,6 +22,8 @@
#include <chrono>
#include <filesystem>
#include <memory>
+#include <tuple>
+#include <vector>
#include "agent/be_exec_version_manager.h"
#include "core/block/block.h"
@@ -60,8 +62,126 @@ PBlock serialize_block(const Block& block) {
return pblock;
}
+void expect_wal_column(const IColumn& column, const std::string& name, size_t
expected_rows,
+ size_t offset) {
+ ASSERT_EQ(column.size(), expected_rows);
+ for (size_t row = 0; row < expected_rows; ++row) {
+ EXPECT_EQ(column.get_data_at(row).to_string(), name +
std::to_string(offset + row));
+ }
+}
+
+void expect_wal_batch(WalReader* reader, Block* output, size_t
expected_columns,
+ size_t expected_rows, size_t offset) {
+ output->clear_column_data();
+ size_t rows = 0;
+ bool eof = true;
+ ASSERT_TRUE(reader->get_block(output, &rows, &eof).ok());
+ // Projection can remove every output column while the WAL block still
contributes rows.
+ EXPECT_EQ(rows, expected_rows);
+ EXPECT_FALSE(eof);
+ ASSERT_EQ(output->columns(), expected_columns);
+ for (const auto& column : output->get_columns_with_type_and_name()) {
+ ASSERT_NO_FATAL_FAILURE(
+ expect_wal_column(*column.column, column.name, expected_rows,
offset));
+ }
+}
+
+void initialize_wal_reader(WalReader* reader, const std::string& path, bool
discover_schema) {
+ reader->_wal_reader = std::make_shared<WalFileReader>(path);
+ ASSERT_TRUE(reader->_wal_reader->init().ok());
+ std::string encoded_ids;
+ ASSERT_TRUE(reader->_wal_reader->read_header(reader->_version,
encoded_ids).ok());
+ ASSERT_TRUE(parse_wal_column_ids(encoded_ids, &reader->_column_ids).ok());
+ if (discover_schema) {
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader->get_schema(&schema).ok());
+ ASSERT_EQ(schema.size(), 2);
+ }
+}
+
+void expect_wal_eof(WalReader* reader, Block* output) {
+ output->clear_column_data();
+ size_t rows = 99;
+ bool eof = false;
+ ASSERT_TRUE(reader->get_block(output, &rows, &eof).ok());
+ EXPECT_EQ(rows, 0);
+ EXPECT_TRUE(eof);
+ EXPECT_EQ(output->rows(), 0);
+}
+
+class WalReaderV2ProjectionTest
+ : public testing::TestWithParam<std::tuple<std::vector<int>, bool>> {
+protected:
+ void SetUp() override {
+ _wal_path = temporary_wal_path();
+ const auto type = std::make_shared<DataTypeString>();
+ WalWriter writer(_wal_path);
+ ASSERT_TRUE(writer.init(io::global_local_filesystem()).ok());
+ ASSERT_TRUE(writer.append_header("17,4").ok());
+ size_t offset = 0;
+ for (const size_t batch_rows : {3, 2}) {
+ Block source;
+ for (const auto& name : {"left_value", "right_value"}) {
+ auto column = type->create_column();
+ for (size_t row = 0; row < batch_rows; ++row) {
+ const auto value = std::string(name) +
std::to_string(offset + row);
+ column->insert_data(value.data(), value.size());
+ }
+ source.insert({std::move(column), type, name});
+ }
+ auto pblock = serialize_block(source);
+ ASSERT_TRUE(writer.append_blocks({&pblock}).ok());
+ offset += batch_rows;
+ }
+ ASSERT_TRUE(writer.finalize().ok());
+ }
+
+ void TearDown() override {
+ std::filesystem::remove_all(
+
std::filesystem::path(_wal_path).parent_path().parent_path().parent_path());
+ }
+
+ std::string _wal_path;
+};
+
} // namespace
+TEST_P(WalReaderV2ProjectionTest, PreservesRowsAcrossBlocksAndEof) {
+ const auto& [projected_columns, discover_schema] = GetParam();
+ std::shared_ptr<io::FileSystemProperties> properties;
+ std::unique_ptr<io::FileDescription> description;
+ WalReader reader(properties, description, nullptr, nullptr, {});
+ ASSERT_NO_FATAL_FAILURE(initialize_wal_reader(&reader, _wal_path,
discover_schema));
+
+ auto request = std::make_shared<FileScanRequest>();
+ Block output;
+ const auto type = std::make_shared<DataTypeString>();
+ const std::vector<std::string> names {"left_value", "right_value"};
+ for (size_t position = 0; position < projected_columns.size(); ++position)
{
+ const auto source_position = projected_columns[position];
+ request->local_positions.emplace(LocalColumnId(source_position),
LocalIndex(position));
+ output.insert({type->create_column(), type, names[source_position]});
+ }
+ ASSERT_TRUE(reader.open(std::move(request)).ok());
+
+ size_t offset = 0;
+ for (const size_t expected_rows : {3, 2}) {
+ ASSERT_NO_FATAL_FAILURE(expect_wal_batch(&reader, &output,
projected_columns.size(),
+ expected_rows, offset));
+ offset += expected_rows;
+ }
+ for (int attempt = 0; attempt < 2; ++attempt) {
+ ASSERT_NO_FATAL_FAILURE(expect_wal_eof(&reader, &output));
+ }
+ ASSERT_TRUE(reader.close().ok());
+}
+
+INSTANTIATE_TEST_SUITE_P(WriterBacked, WalReaderV2ProjectionTest,
+ testing::Combine(testing::Values(std::vector<int> {},
std::vector<int> {0},
+ std::vector<int> {1},
+ std::vector<int> {1,
0}),
+ testing::Bool()));
+
TEST(WalReaderV2Test, ParseColumnIdsPreservesHeaderOrder) {
std::vector<int32_t> column_ids;
ASSERT_TRUE(parse_wal_column_ids("17,4,99", &column_ids).ok());
diff --git a/contrib/apache-orc b/contrib/apache-orc
index be0f1b73a7a..5962bb071d6 160000
--- a/contrib/apache-orc
+++ b/contrib/apache-orc
@@ -1 +1 @@
-Subproject commit be0f1b73a7aeb78824a03e0dcb692c50a176d513
+Subproject commit 5962bb071d6345cc29c5353d64d85a203525e4d9
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index 7f9f773133f..f7f6909e4a1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -345,7 +345,6 @@ import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
@@ -363,6 +362,7 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.ObjLongConsumer;
import java.util.stream.Collectors;
// Frontend service used to serve all request for this frontend through
@@ -496,7 +496,7 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
List<Long> dbIds = Lists.newArrayList();
List<Long> catalogIds = Lists.newArrayList();
- PatternMatcher matcher = null;
+ PatternMatcher matcher;
if (params.isSetPattern()) {
try {
matcher =
PatternMatcher.createMysqlPattern(params.getPattern(),
@@ -504,6 +504,8 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
} catch (PatternMatcherException e) {
throw new TException("Pattern is in bad format: " +
params.getPattern());
}
+ } else {
+ matcher = null;
}
Env env = Env.getCurrentEnv();
@@ -517,55 +519,77 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
catalog -> new TException("Unknown catalog " +
catalog)));
}
+ boolean includeDbIds = params.isIncludeDbIds();
for (CatalogIf catalog : catalogIfs) {
- Collection<DatabaseIf> dbs = new HashSet<DatabaseIf>();
+ Collection<DatabaseIf> dbs = Collections.emptyList();
+ List<String> names = Collections.emptyList();
try {
- dbs = catalog.getAllDbs();
+ if (includeDbIds) {
+ dbs = catalog.getAllDbs();
+ } else {
+ // SCHEMATA only needs names; loading every database
churns the object cache on large catalogs.
+ names = catalog.getDbNames();
+ }
} catch (Exception e) {
LOG.warn("failed to get database names for catalog {}",
catalog.getName(), e);
// Some external catalog may fail to get databases due to
wrong connection info.
}
+ int dbCount = includeDbIds ? dbs.size() : names.size();
if (LOG.isDebugEnabled()) {
- LOG.debug("get db size: {}, in catalog: {}", dbs.size(),
catalog.getName());
+ LOG.debug("get db size: {}, in catalog: {}", dbCount,
catalog.getName());
}
- if (dbs.isEmpty() && params.isSetGetNullCatalog() &&
params.get_null_catalog) {
+ if (dbCount == 0 && params.isSetGetNullCatalog() &&
params.get_null_catalog) {
catalogNames.add(catalog.getName());
dbNames.add("NULL");
catalogIds.add(catalog.getId());
- dbIds.add(-1L);
+ if (includeDbIds) {
+ dbIds.add(-1L);
+ }
continue;
}
- if (dbs.isEmpty()) {
+ if (dbCount == 0) {
continue;
}
- UserIdentity currentUser = null;
+ UserIdentity currentUser;
if (params.isSetCurrentUserIdent()) {
currentUser =
UserIdentity.fromThrift(params.current_user_ident);
} else {
currentUser =
UserIdentity.createAnalyzedUserIdentWithIp(params.user, params.user_ip);
}
- for (DatabaseIf db : dbs) {
- String dbName = db.getFullName();
+ ObjLongConsumer<String> addDatabase = (dbName, dbId) -> {
if (!env.getAccessManager()
.checkDbPriv(currentUser, catalog.getName(), dbName,
PrivPredicate.SHOW)) {
- continue;
+ return;
}
if (matcher != null &&
!matcher.match(getMysqlTableSchema(catalog.getName(), dbName))) {
- continue;
+ return;
}
catalogNames.add(catalog.getName());
dbNames.add(getMysqlTableSchema(catalog.getName(), dbName));
catalogIds.add(catalog.getId());
- dbIds.add(db.getId());
+ if (includeDbIds) {
+ dbIds.add(dbId);
+ }
+ };
+ if (includeDbIds) {
+ for (DatabaseIf db : dbs) {
+ addDatabase.accept(db.getFullName(), db.getId());
+ }
+ } else {
+ for (String dbName : names) {
+ addDatabase.accept(dbName, -1L);
+ }
}
}
result.setDbs(dbNames);
result.setCatalogs(catalogNames);
result.setCatalogIds(catalogIds);
- result.setDbIds(dbIds);
+ if (includeDbIds) {
+ result.setDbIds(dbIds);
+ }
return result;
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceGetDbNamesTest.java
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceGetDbNamesTest.java
new file mode 100644
index 00000000000..236569257fd
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceGetDbNamesTest.java
@@ -0,0 +1,186 @@
+// 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.
+
+package org.apache.doris.service;
+
+import org.apache.doris.analysis.UserIdentity;
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.InfoSchemaDb;
+import org.apache.doris.datasource.CatalogIf;
+import org.apache.doris.datasource.CatalogMgr;
+import org.apache.doris.mysql.privilege.AccessControllerManager;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.GlobalVariable;
+import org.apache.doris.thrift.TGetDbsParams;
+import org.apache.doris.thrift.TGetDbsResult;
+
+import org.apache.thrift.TDeserializer;
+import org.apache.thrift.TSerializer;
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+public class FrontendServiceGetDbNamesTest {
+ private final Env env = Mockito.mock(Env.class);
+ private final CatalogMgr catalogMgr = Mockito.mock(CatalogMgr.class);
+ private final AccessControllerManager accessManager =
Mockito.mock(AccessControllerManager.class);
+ private final CatalogIf catalog = Mockito.mock(CatalogIf.class);
+ private final FrontendServiceImpl service = new
FrontendServiceImpl(Mockito.mock(ExecuteEnv.class));
+ private MockedStatic<Env> envMock;
+
+ @BeforeEach
+ public void setUp() {
+ envMock = Mockito.mockStatic(Env.class);
+ envMock.when(Env::getCurrentEnv).thenReturn(env);
+ Mockito.when(env.getCatalogMgr()).thenReturn(catalogMgr);
+ Mockito.when(env.getAccessManager()).thenReturn(accessManager);
+
Mockito.when(catalogMgr.listCatalogs()).thenReturn(Collections.singletonList(catalog));
+ Mockito.when(catalog.getName()).thenReturn("example_catalog");
+ Mockito.when(catalog.getId()).thenReturn(42L);
+
Mockito.when(accessManager.checkDbPriv(Mockito.any(UserIdentity.class),
Mockito.anyString(),
+ Mockito.anyString(),
Mockito.eq(PrivPredicate.SHOW))).thenReturn(true);
+ }
+
+ @AfterEach
+ public void tearDown() {
+ envMock.close();
+ }
+
+ @Test
+ public void testNamesOnlyDoesNotMaterializeDatabases() throws Exception {
+ List<String> names = new ArrayList<>();
+ for (int i = 0; i < 100_000; i++) {
+ names.add("namespace_" + i);
+ }
+ names.add(InfoSchemaDb.DATABASE_NAME);
+ names.add("mysql");
+ Mockito.when(catalog.getDbNames()).thenReturn(names);
+ Mockito.when(catalog.getAllDbs()).thenThrow(new
AssertionError("Database objects must not be loaded"));
+
+ TGetDbsParams params = params();
+ params.setIncludeDbIds(false);
+ TGetDbsResult result = service.getDbNames(params);
+
+ Assertions.assertEquals(names.size(), result.getDbsSize());
+ Assertions.assertEquals(names.size(), result.getCatalogsSize());
+ Assertions.assertEquals(names.size(), result.getCatalogIdsSize());
+
Assertions.assertTrue(result.getCatalogs().stream().allMatch("example_catalog"::equals));
+ Assertions.assertTrue(result.getCatalogIds().stream().allMatch(id ->
id == 42L));
+ Assertions.assertFalse(result.isSetDbIds());
+ Mockito.verify(catalog, Mockito.times(1)).getDbNames();
+ Mockito.verify(catalog, Mockito.never()).getAllDbs();
+ }
+
+ @Test
+ public void testNamesOnlyPreservesPrivilegesPatternAndDisplayNames()
throws Exception {
+
Mockito.when(catalog.getDbNames()).thenReturn(Arrays.asList("visible_a",
"visible_hidden", "other"));
+
Mockito.when(accessManager.checkDbPriv(Mockito.any(UserIdentity.class),
Mockito.eq("example_catalog"),
+ Mockito.eq("visible_hidden"),
Mockito.eq(PrivPredicate.SHOW))).thenReturn(false);
+
+ boolean original = GlobalVariable.showFullDbNameInInfoSchemaDb;
+ try {
+ for (boolean showFullName : new boolean[] {false, true}) {
+ GlobalVariable.showFullDbNameInInfoSchemaDb = showFullName;
+ TGetDbsParams params = params();
+ params.setIncludeDbIds(false);
+ params.setPattern(showFullName ? "example_catalog.visible%" :
"visible%");
+ TGetDbsResult result = service.getDbNames(params);
+ Assertions.assertEquals(Collections.singletonList(
+ showFullName ? "example_catalog.visible_a" :
"visible_a"), result.getDbs());
+ Assertions.assertFalse(result.isSetDbIds());
+ }
+ } finally {
+ GlobalVariable.showFullDbNameInInfoSchemaDb = original;
+ }
+ }
+
+ @Test
+ public void testExistingCallersKeepDatabaseIds() throws Exception {
+ DatabaseIf database = Mockito.mock(DatabaseIf.class);
+ Mockito.when(database.getFullName()).thenReturn("visible_a");
+ Mockito.when(database.getId()).thenReturn(123L);
+
Mockito.when(catalog.getAllDbs()).thenReturn(Collections.singletonList(database));
+
+ for (boolean explicitlySet : new boolean[] {false, true}) {
+ TGetDbsParams params = params();
+ if (explicitlySet) {
+ params.setIncludeDbIds(true);
+ }
+ TGetDbsResult result = service.getDbNames(params);
+ Assertions.assertEquals(Collections.singletonList(123L),
result.getDbIds());
+ Assertions.assertEquals(Collections.singletonList(42L),
result.getCatalogIds());
+ }
+ Mockito.verify(catalog, Mockito.never()).getDbNames();
+ }
+
+ @Test
+ public void testAbsentIncludeDbIdsOnWireKeepsLegacyDefault() throws
Exception {
+ TGetDbsParams legacy = params();
+ legacy.unsetIncludeDbIds();
+ TSerializer serializer = new TSerializer(new
TBinaryProtocol.Factory());
+ TDeserializer deserializer = new TDeserializer(new
TBinaryProtocol.Factory());
+ TGetDbsParams decoded = new TGetDbsParams();
+ deserializer.deserialize(decoded, serializer.serialize(legacy));
+ Assertions.assertTrue(decoded.isIncludeDbIds());
+
+ legacy.setIncludeDbIds(false);
+ deserializer.deserialize(decoded, serializer.serialize(legacy));
+ Assertions.assertFalse(decoded.isIncludeDbIds());
+ }
+
+ @Test
+ public void testNamesOnlyPreservesEmptyCatalogPlaceholder() throws
Exception {
+ Mockito.when(catalog.getDbNames()).thenReturn(Collections.emptyList());
+ TGetDbsParams params = params();
+ params.setIncludeDbIds(false);
+ params.setGetNullCatalog(true);
+
+ TGetDbsResult result = service.getDbNames(params);
+ Assertions.assertEquals(Collections.singletonList("NULL"),
result.getDbs());
+ Assertions.assertEquals(Collections.singletonList("example_catalog"),
result.getCatalogs());
+ Assertions.assertEquals(Collections.singletonList(42L),
result.getCatalogIds());
+ Assertions.assertFalse(result.isSetDbIds());
+ }
+
+ @Test
+ public void testNamesOnlySkipsUnavailableCatalog() throws Exception {
+ Mockito.when(catalog.getDbNames()).thenThrow(new
IllegalStateException("Metadata service unavailable"));
+ TGetDbsParams params = params();
+ params.setIncludeDbIds(false);
+
+ TGetDbsResult result = service.getDbNames(params);
+ Assertions.assertTrue(result.getDbs().isEmpty());
+ Assertions.assertTrue(result.getCatalogs().isEmpty());
+ Assertions.assertFalse(result.isSetDbIds());
+ }
+
+ private TGetDbsParams params() {
+ TGetDbsParams params = new TGetDbsParams();
+ params.setCurrentUserIdent(UserIdentity.ROOT.toThrift());
+ return params;
+ }
+}
diff --git a/gensrc/thrift/FrontendService.thrift
b/gensrc/thrift/FrontendService.thrift
index 4e753eb859b..d1624f6ac2e 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -101,6 +101,8 @@ struct TGetDbsParams {
// Reserved for downstream field `current_roles` to keep thrift field ids
// wire-compatible across maintained branches. Do not reuse this id.
7: optional set<string> reserved_field_7
+ // Name-only scanners can avoid materializing database objects and omit
db_ids in the response.
+ 8: optional bool include_db_ids = true
}
// getDbNames returns a list of database names , database ids and catalog
names ,catalog ids
diff --git
a/regression-test/data/external_table_p0/tvf/test_data/GenerateOrcConditionCacheSeek.cpp
b/regression-test/data/external_table_p0/tvf/test_data/GenerateOrcConditionCacheSeek.cpp
new file mode 100644
index 00000000000..be4ec1bbb61
--- /dev/null
+++
b/regression-test/data/external_table_p0/tvf/test_data/GenerateOrcConditionCacheSeek.cpp
@@ -0,0 +1,51 @@
+// 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 <memory>
+#include <orc/OrcFile.hh>
+#include <orc/Writer.hh>
+
+// Link against the ORC library used by Doris and pass the output fixture path.
+// Each stripe has 20,000 rows and two 10,000-row index groups. Only its first
row
+// matches id > 5000. The second match makes cache granule [18432, 20480) true,
+// so a cache hit seeks into the first stripe's SARG-filtered suffix at row
18432.
+int main(int argc, char** argv) {
+ if (argc != 2) {
+ return 1;
+ }
+ auto stream = orc::writeLocalFile(argv[1]);
+ auto type = std::unique_ptr<orc::Type>(
+ orc::Type::buildTypeFromString("struct<id:int,payload:bigint>"));
+ orc::WriterOptions options;
+
options.setStripeSize(1).setRowIndexStride(10000).setCompression(orc::CompressionKind_NONE);
+ auto writer = orc::createWriter(*type, stream.get(), options);
+ auto batch = writer->createRowBatch(20000);
+ auto& root = dynamic_cast<orc::StructVectorBatch&>(*batch);
+ auto& ids = dynamic_cast<orc::LongVectorBatch&>(*root.fields[0]);
+ auto& payloads = dynamic_cast<orc::LongVectorBatch&>(*root.fields[1]);
+ for (int64_t stripe = 0; stripe < 2; ++stripe) {
+ for (int64_t row = 0; row < 20000; ++row) {
+ ids.data[row] = row == 0 ? 10000 + stripe : 0;
+ payloads.data[row] = stripe * 20000 + row;
+ }
+ root.numElements = ids.numElements = payloads.numElements = 20000;
+ writer->add(*batch);
+ }
+ writer->close();
+ stream->close();
+ return 0;
+}
diff --git
a/regression-test/data/external_table_p0/tvf/test_data/condition_cache_seek.orc
b/regression-test/data/external_table_p0/tvf/test_data/condition_cache_seek.orc
new file mode 100644
index 00000000000..3cb2b0524c0
Binary files /dev/null and
b/regression-test/data/external_table_p0/tvf/test_data/condition_cache_seek.orc
differ
diff --git
a/regression-test/data/insert_p0/group_commit/test_group_commit_wal_row_count.out
b/regression-test/data/insert_p0/group_commit/test_group_commit_wal_row_count.out
new file mode 100644
index 00000000000..0ae4de99333
--- /dev/null
+++
b/regression-test/data/insert_p0/group_commit/test_group_commit_wal_row_count.out
@@ -0,0 +1,17 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !before_replay --
+0
+
+-- !counts --
+5 4 15 120
+
+-- !values --
+1 10
+2 20
+3 \N
+4 40
+5 50
+
+-- !filtered_count --
+3
+
diff --git
a/regression-test/suites/external_table_p0/hive/test_hive_json_partition_only.groovy
b/regression-test/suites/external_table_p0/hive/test_hive_json_partition_only.groovy
new file mode 100644
index 00000000000..69ae572d563
--- /dev/null
+++
b/regression-test/suites/external_table_p0/hive/test_hive_json_partition_only.groovy
@@ -0,0 +1,85 @@
+// 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.
+
+suite("test_hive_json_partition_only", "p0,external,hive") {
+ String enabled = context.config.otherConfigs.get("enableHiveTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable Hive test")
+ return
+ }
+
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String hmsPort = context.config.otherConfigs.get("hive3HmsPort")
+ String catalog = "test_hive_json_partition_only"
+ def previousSettings = sql """SELECT
@@enable_count_push_down_for_external_table,
+ @@enable_sql_cache"""
+ sql """DROP CATALOG IF EXISTS ${catalog}"""
+ try {
+ sql """CREATE CATALOG ${catalog} PROPERTIES (
+ "type" = "hms",
+ "hive.metastore.uris" = "thrift://${externalEnvIp}:${hmsPort}"
+ )"""
+ sql """USE `${catalog}`.`default`"""
+ sql "SET enable_sql_cache=false"
+
+ // Reading a real JSON field provides the row-count oracle for the
same partition files.
+ def baseline = sql "SELECT id, dt FROM json_all_complex_types"
+ assertTrue(!baseline.isEmpty())
+ def partitions = baseline.groupBy { row -> row[1] }
+
+ for (boolean pushdown : [false, true]) {
+ sql "SET enable_count_push_down_for_external_table=${pushdown}"
+ partitions.each { partition, rows ->
+ assertNotNull(partition)
+ String value = partition.toString().replace("'", "''")
+ String predicate = "dt = '${value}'"
+ String countQuery = "SELECT COUNT(*) FROM
json_all_complex_types WHERE ${predicate}"
+ explain {
+ sql countQuery
+ contains "pushdown agg=NONE"
+ }
+ def count = sql countQuery
+ assertEquals(rows.size(), (count[0][0] as Number).intValue())
+
+ // Partition values come from the split, so neither query
needs a JSON file column.
+ def projected = sql "SELECT dt FROM json_all_complex_types
WHERE ${predicate}"
+ assertEquals(rows.size(), projected.size())
+ assertTrue(projected.every { row -> row[0] == partition })
+ def partitionCount = sql "SELECT COUNT(dt) FROM
json_all_complex_types WHERE ${predicate}"
+ assertEquals(rows.size(), (partitionCount[0][0] as
Number).intValue())
+
+ def filtered = sql """SELECT COUNT(*) FROM
json_all_complex_types
+ WHERE ${predicate} AND id IS NOT NULL"""
+ assertEquals(rows.count { row -> row[0] != null },
+ (filtered[0][0] as Number).intValue())
+ }
+
+ def grouped = sql "SELECT dt, COUNT(*) FROM json_all_complex_types
GROUP BY dt"
+ assertEquals(partitions.size(), grouped.size())
+ grouped.each { row ->
+ assertEquals(partitions[row[0]].size(), (row[1] as
Number).intValue())
+ }
+ def empty = sql """SELECT COUNT(*) FROM json_all_complex_types
+ WHERE dt = '__missing_json_partition__'"""
+ assertEquals(0, (empty[0][0] as Number).intValue())
+ }
+ } finally {
+ sql "SET
enable_count_push_down_for_external_table=${previousSettings[0][0]}"
+ sql "SET enable_sql_cache=${previousSettings[0][1]}"
+ sql """DROP CATALOG IF EXISTS ${catalog}"""
+ }
+}
diff --git
a/regression-test/suites/external_table_p0/tvf/test_orc_condition_cache_seek.groovy
b/regression-test/suites/external_table_p0/tvf/test_orc_condition_cache_seek.groovy
new file mode 100644
index 00000000000..e6a056f5747
--- /dev/null
+++
b/regression-test/suites/external_table_p0/tvf/test_orc_condition_cache_seek.groovy
@@ -0,0 +1,110 @@
+// 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.
+
+import com.azure.storage.blob.BlobClientBuilder
+import com.azure.storage.common.StorageSharedKeyCredential
+import org.apache.doris.regression.action.ProfileAction
+
+suite("test_orc_condition_cache_seek", "p0,external") {
+ String provider = getS3Provider()
+ String endpoint = getS3Endpoint()
+ if (provider.equalsIgnoreCase("AZURE")) {
+ if (endpoint == "azure-endpoint") {
+ endpoint = "https://${getS3AK()}.blob.core.windows.net"
+ } else if (!endpoint.contains("://")) {
+ endpoint = "https://${endpoint}";
+ }
+ }
+ String bucket = context.config.otherConfigs.get("s3BucketName")
+ String key = "orc_condition_cache_seek/${UUID.randomUUID()}/values.orc"
+ File fixture = new File(context.dataPath,
"test_data/condition_cache_seek.orc")
+ def blobClient = null
+ def s3Client = null
+ boolean uploaded = false
+ Map<String, String> originalSettings = [:]
+ for (String variable : ["enable_condition_cache",
"enable_orc_lazy_materialization",
+ "enable_orc_filter_by_min_max", "batch_size",
"enable_sql_cache",
+ "enable_query_cache", "enable_profile",
"profile_level", "parallel_pipeline_task_num",
+ "max_file_scanners_concurrency"]) {
+ originalSettings[variable] = sql("SHOW VARIABLES LIKE
'${variable}'")[0][1].toString()
+ }
+ try {
+ if (provider.equalsIgnoreCase("AZURE")) {
+ blobClient = new
BlobClientBuilder().endpoint(endpoint).containerName(bucket)
+ .blobName(key).credential(new
StorageSharedKeyCredential(getS3AK(), getS3SK()))
+ .buildClient()
+ blobClient.uploadFromFile(fixture.absolutePath)
+ } else {
+ s3Client = getS3Client()
+ s3Client.putObject(bucket, key, fixture)
+ }
+ uploaded = true
+ String source = """S3('uri'='s3://${bucket}/${key}', 'format'='orc',
+ 'provider'='${provider}', 's3.endpoint'='${endpoint}',
's3.region'='${getS3Region()}',
+ 's3.access_key'='${getS3AK()}',
's3.secret_key'='${getS3SK()}')"""
+ sql "SET enable_sql_cache = false"
+ sql "SET enable_query_cache = false"
+ sql "SET enable_profile = true"
+ sql "SET profile_level = 2"
+ sql "SET parallel_pipeline_task_num = 1"
+ sql "SET max_file_scanners_concurrency = 1"
+ sql "SET batch_size = 2048"
+ // A full LRU-K shard needs one scan for ghost history and another for
admission.
+ // Allow a third visit to one backend even when scans are spread
across all backends.
+ int attempts = 2 * sql("SHOW BACKENDS").size() + 1
+ for (boolean minMax : [false, true]) {
+ sql "SET enable_orc_filter_by_min_max = ${minMax}"
+ for (boolean lazy : [false, true]) {
+ sql "SET enable_orc_lazy_materialization = ${lazy}"
+ sql "SET enable_condition_cache = false"
+ quickTest("uncached", "SELECT id, payload FROM ${source} WHERE
id > 5000 ORDER BY id",
+ "SELECT 10000, CAST(0 AS BIGINT) UNION ALL SELECT
10001, CAST(20000 AS BIGINT) ORDER BY 1")
+ sql "SET enable_condition_cache = true"
+ boolean cacheHit = false
+ for (int attempt = 0; attempt < attempts && !cacheHit;
++attempt) {
+ String token =
"orc_seek_${UUID.randomUUID().toString().replace('-', '')}"
+ // The first match populates granule 0. The next stripe's
match populates
+ // granule 9, whose start (18432) lies in the first
stripe's pruned suffix.
+ quickTest("cached",
+ "SELECT id, payload FROM (SELECT id, payload,
'${token}' AS marker " +
+ "FROM ${source} WHERE id > 5000) t ORDER
BY id",
+ "SELECT 10000, CAST(0 AS BIGINT) UNION ALL " +
+ "SELECT 10001, CAST(20000 AS BIGINT) ORDER
BY 1")
+ String profile = new
ProfileAction(context).getProfileBySql(token,
+ ["ConditionCacheHit",
"ConditionCacheFilteredRows"])
+ cacheHit = (profile =~
/(?m)ConditionCacheHit:\s*[1-9]/).find()
+ if (cacheHit) {
+ assertTrue((profile =~
/(?m)ConditionCacheFilteredRows:\s*[1-9]/).find(), profile)
+ }
+ }
+ assertTrue(cacheHit, "No condition cache hit after ${attempts}
scans")
+ }
+ }
+ } finally {
+ try {
+ originalSettings.each { variable, value -> sql "SET ${variable} =
${value}" }
+ } finally {
+ if (uploaded) {
+ if (blobClient != null) {
+ blobClient.delete()
+ } else {
+ s3Client.deleteObject(bucket, key)
+ }
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/insert_p0/group_commit/test_group_commit_wal_row_count.groovy
b/regression-test/suites/insert_p0/group_commit/test_group_commit_wal_row_count.groovy
new file mode 100644
index 00000000000..ed1a449724a
--- /dev/null
+++
b/regression-test/suites/insert_p0/group_commit/test_group_commit_wal_row_count.groovy
@@ -0,0 +1,85 @@
+// 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.
+
+import org.awaitility.Awaitility
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_group_commit_wal_row_count", "nonConcurrent") {
+ sql "DROP TABLE IF EXISTS test_group_commit_wal_row_count"
+ sql """
+ CREATE TABLE test_group_commit_wal_row_count (
+ k INT NOT NULL,
+ v INT NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1",
+ "group_commit_interval_ms" = "2000"
+ )
+ """
+
+ String loadError = "LoadBlockQueue._finish_group_commit_load.load_error"
+ String replayStop = "WalTable.replay_wals.stop"
+ try {
+ GetDebugPoint().enableDebugPointForAllBEs(loadError)
+ GetDebugPoint().enableDebugPointForAllBEs(replayStop)
+
+ // Abort each original transaction before the next load so both WALs
must be replayed.
+ for (String data : ["1,10\n2,20\n3,\\N\n", "4,40\n5,50\n"]) {
+ long txnId = 0
+ streamLoad {
+ table "test_group_commit_wal_row_count"
+ set "column_separator", ","
+ set "group_commit", "async_mode"
+ unset "label"
+ inputStream new ByteArrayInputStream(data.getBytes("UTF-8"))
+ time 10000
+ check { result, exception, startTime, endTime ->
+ assertNull(exception)
+ def response = parseJson(result)
+ assertEquals("Success", response.Status)
+ assertEquals(data.readLines().size(),
response.NumberTotalRows as int)
+ txnId = response.TxnId as long
+ assertTrue(txnId > 0)
+ }
+ }
+ Awaitility.await().atMost(30, SECONDS).pollInterval(1,
SECONDS).until {
+ def txn = sql_return_maparray "SHOW TRANSACTION WHERE ID =
${txnId}"
+ return txn.size() == 1 && txn[0].TransactionStatus == "ABORTED"
+ }
+ }
+ qt_before_replay "SELECT COUNT(*) FROM test_group_commit_wal_row_count"
+
+ // Keep the original-load failure enabled: rows can become visible
only through WAL replay.
+ GetDebugPoint().disableDebugPointForAllBEs(replayStop)
+ Awaitility.await().atMost(90, SECONDS).pollInterval(1, SECONDS).until {
+ def rows = sql "SELECT COUNT(*) FROM
test_group_commit_wal_row_count"
+ return rows[0][0] == 5
+ }
+ qt_counts """SELECT COUNT(*), COUNT(v), SUM(k), SUM(v)
+ FROM test_group_commit_wal_row_count"""
+ qt_values "SELECT k, v FROM test_group_commit_wal_row_count ORDER BY k"
+ qt_filtered_count "SELECT COUNT(*) FROM
test_group_commit_wal_row_count WHERE v >= 20"
+ } finally {
+ try {
+ GetDebugPoint().disableDebugPointForAllBEs(replayStop)
+ } finally {
+ GetDebugPoint().disableDebugPointForAllBEs(loadError)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
