github-actions[bot] commented on code in PR #66685:
URL: https://github.com/apache/doris/pull/66685#discussion_r3768699279
##########
be/src/common/config.cpp:
##########
@@ -1581,9 +1581,10 @@ DEFINE_mInt64(hive_sink_max_file_size, "1073741824"); //
1GB
DEFINE_mInt64(iceberg_sink_max_file_size, "1073741824"); // 1GB
/** Paimon sink configurations **/
-DEFINE_mInt64(paimon_jni_writer_memory_pool_limit_bytes, "536870912"); // 512MB
-DEFINE_Validator(paimon_jni_writer_memory_pool_limit_bytes,
- [](int64_t bytes) -> bool { return bytes > 0; });
+// Process-wide hard limit as a fraction of the BE JVM -Xmx.
+DEFINE_mDouble(paimon_jni_memory_limit_ratio, "0.5");
+DEFINE_Validator(paimon_jni_memory_limit_ratio,
Review Comment:
[P2] Validate the requested mutable ratio
This is a mutable config, but `UPDATE_FIELD` calls the registered validator
before assigning `new_value`; the generated zero-argument validator therefore
reads the old global ratio. Starting from 0.5, `set_config(..., "2")` (and
likewise a negative value or NaN) succeeds and only then stores the invalid
value, defeating the `(0, 1]` hard-limit invariant. Please either make this
setting immutable or validate the proposed value before committing it, with a
live-update regression case.
##########
be/src/exprs/function/function_paimon_routing.cpp:
##########
@@ -0,0 +1,203 @@
+// 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 <cstdint>
+#include <cstring>
+#include <limits>
+#include <memory>
+#include <optional>
+#include <string_view>
+#include <utility>
+
+#include "core/block/block.h"
+#include "core/data_type/data_type_nullable.h"
+#include "core/data_type/data_type_number.h"
+#include "exec/sink/paimon_native_row_hash.h"
+#include "exprs/function/function.h"
+#include "exprs/function/simple_function_factory.h"
+
+namespace doris {
+namespace {
+
+template <typename T>
+bool read_fixed_value(const IColumn& column, size_t row, T* value) {
+ StringRef data = column.get_data_at(row);
+ if (data.size != sizeof(T)) {
+ return false;
+ }
+ std::memcpy(value, data.data, sizeof(T));
+ return true;
+}
+
+Status encode_field(paimon_native::BinaryRowEncoder* encoder, size_t
target_position,
+ const ColumnWithTypeAndName& field, size_t row) {
+ const IColumn& column = *field.column;
+ if (column.is_null_at(row)) {
+ if (!encoder->set_null(target_position)) {
+ return Status::InternalError("Failed to encode null Paimon routing
field {}",
+ target_position);
+ }
+ return Status::OK();
+ }
+
+ bool encoded = false;
+ switch (remove_nullable(field.type)->get_primitive_type()) {
+ case TYPE_BOOLEAN: {
+ uint8_t value = 0;
+ encoded = read_fixed_value(column, row, &value) &&
+ encoder->write_boolean(target_position, value != 0);
+ break;
+ }
+ case TYPE_TINYINT: {
+ int8_t value = 0;
+ encoded = read_fixed_value(column, row, &value) &&
+ encoder->write_tinyint(target_position, value);
+ break;
+ }
+ case TYPE_SMALLINT: {
+ int16_t value = 0;
+ encoded = read_fixed_value(column, row, &value) &&
+ encoder->write_smallint(target_position, value);
+ break;
+ }
+ case TYPE_INT: {
+ int32_t value = 0;
+ encoded =
+ read_fixed_value(column, row, &value) &&
encoder->write_int(target_position, value);
+ break;
+ }
+ case TYPE_BIGINT: {
+ int64_t value = 0;
+ encoded = read_fixed_value(column, row, &value) &&
+ encoder->write_bigint(target_position, value);
+ break;
+ }
+ case TYPE_FLOAT: {
+ float value = 0;
+ encoded = read_fixed_value(column, row, &value) &&
+ encoder->write_float(target_position, value);
+ break;
+ }
+ case TYPE_DOUBLE: {
+ double value = 0;
+ encoded = read_fixed_value(column, row, &value) &&
+ encoder->write_double(target_position, value);
+ break;
+ }
+ case TYPE_CHAR:
+ case TYPE_VARCHAR:
+ case TYPE_STRING: {
+ StringRef value = column.get_data_at(row);
+ encoded = encoder->write_string(target_position,
std::string_view(value.data, value.size));
+ break;
+ }
+ case TYPE_BINARY:
+ case TYPE_VARBINARY: {
+ StringRef value = column.get_data_at(row);
+ encoded = encoder->write_binary(target_position,
std::string_view(value.data, value.size));
+ break;
+ }
+ default:
+ return Status::InvalidArgument("Unsupported Doris type {} for Paimon
native routing",
+ field.type->get_name());
+ }
+ if (!encoded) {
+ return Status::InvalidArgument("Doris column {} cannot be encoded for
Paimon routing",
+ field.name);
+ }
+ return Status::OK();
+}
+
+enum class PaimonRoutingResult { BINARY_ROW_HASH, FIXED_BUCKET };
+
+template <PaimonRoutingResult result_kind>
+class FunctionPaimonRouting final : public IFunction {
+public:
+ static constexpr auto name = result_kind ==
PaimonRoutingResult::BINARY_ROW_HASH
+ ? "__paimon_binary_row_hash_v1"
+ : "__paimon_fixed_bucket_v1";
+
+ static FunctionPtr create() { return
std::make_shared<FunctionPaimonRouting<result_kind>>(); }
+
+ String get_name() const override { return name; }
+
+ bool is_variadic() const override { return true; }
+
+ size_t get_number_of_arguments() const override { return 0; }
+
+ bool use_default_implementation_for_nulls() const override { return false;
}
+
+ ColumnNumbers get_arguments_that_are_always_constant() const override {
Review Comment:
[P1] Avoid the all-constant fixed-bucket out-of-bounds path
When all fixed-bucket key expressions are constant on a multi-row input (for
example, `INSERT ... SELECT 1, payload FROM source`), the default
constant-argument wrapper retains argument 0 as an N-row `ColumnConst` but
unwraps every key to a one-row nested column. The temporary block therefore
still reports N rows, and this loop indexes those one-row key columns at rows
1..N-1. Nullable keys first read past the one-byte null map; fixed/string keys
likewise index past their nested column, so a valid insert can crash or
misroute rows. Please disable the generic constant fast path here (the
implementation already handles `ColumnConst`) or otherwise make the temporary
block cardinalities consistent, and cover multi-row constant and constant-NULL
keys.
##########
be/src/exec/sink/writer/paimon/paimon_jni_memory_manager.cpp:
##########
@@ -27,33 +29,101 @@
#include "common/exception.h"
#include "common/logging.h"
#include "core/allocator.h"
-#include "runtime/memory/mem_tracker_limiter.h"
#include "runtime/query_context.h"
#include "runtime/runtime_state.h"
#include "runtime/thread_context.h"
#include "util/defer_op.h"
#include "util/jni-util.h"
-#include "util/pretty_printer.h"
namespace doris {
+namespace {
+
+class ProcessPaimonJniMemoryLimiter {
+public:
+ int64_t limit() const {
+ const long double limit =
+ static_cast<long
double>(Jni::Util::get_max_jni_heap_memory_size()) *
Review Comment:
[P2] Use the real JVM heap on USE_LIBHDFS3 builds
`get_max_jni_heap_memory_size()` returns `SIZE_MAX` whenever `USE_LIBHDFS3`
is defined, so the default ratio makes this expression clamp to `INT64_MAX`.
Those builds still create a finite JVM (`-Xmx1g` by default, or `JAVA_OPTS`),
and this value is passed to both the process limiter and the Arrow allocator;
concurrent Paimon writes therefore have no useful hard cap and can exhaust that
JVM. The ordinary parser also misreads valid unitless `-Xmx` values. Please
derive the limit from the created VM/runtime and cover both a finite
`USE_LIBHDFS3` heap and unitless heap options.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/distribute/worker/job/UnassignedShuffleJob.java:
##########
@@ -87,6 +87,12 @@ protected int degreeOfParallelism() {
if (connectContext != null && connectContext.getSessionVariable() !=
null) {
expectInstanceNum =
connectContext.getSessionVariable().getExchangeInstanceParallel();
}
+ int writerInstanceLimit = fragment.getSink() == null
Review Comment:
[P2] Enforce writer caps in the legacy scheduler too
This is the only scheduling consumer of `getWriterInstanceLimit()`. When
`enable_nereids_distribute_planner=false`, the legacy `Coordinator` instead
copies every instance of the largest child fragment (or uses
`exchange_instance_parallel`) for partitioned exchange-root sinks without
consulting this cap. A four-bucket Paimon plan that reports `planned writers:
4` can therefore still open, for example, 64 JNI writers, defeating the
resource/small-writer control on a supported session path. Please apply the
same upper bound in the legacy partitioned branch and cover that scheduler mode.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalWriterParallelismPolicy.java:
##########
@@ -0,0 +1,127 @@
+// 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.datasource;
+
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.statistics.ColumnStatistic;
+import org.apache.doris.statistics.Statistics;
+
+import java.util.OptionalLong;
+
+/** Plans external writer count without changing ownership-key semantics at
runtime. */
+public final class ExternalWriterParallelismPolicy {
+ private ExternalWriterParallelismPolicy() {
+ }
+
+ /**
+ * Plan writer parallelism. The scheduler may use fewer writers when the
upstream fragment has
+ * less capacity, but must never exceed this result.
+ */
+ public static ExternalWriterParallelism plan(ExternalWriteDistributionPlan
distribution,
+ Statistics statistics, int clusterWriterCapacity) {
+ if (clusterWriterCapacity <= 0) {
+ throw new IllegalArgumentException("writer capacity must be
positive");
+ }
+
+ OptionalLong estimatedOwnershipCount;
+ long planned;
+ if (distribution.isSingleWriter()) {
+ estimatedOwnershipCount = OptionalLong.of(1);
+ planned = 1;
+ } else if (distribution.isRandom()) {
+ estimatedOwnershipCount = OptionalLong.empty();
+ planned = clusterWriterCapacity;
+ } else {
+ estimatedOwnershipCount = estimateOwnershipCount(distribution,
statistics);
+ // Adaptive hash starts with one writer per key but may fan a hot
key out to more
+ // writers. Keep the full fragment capacity available for that
runtime decision.
+ planned = distribution.isAdaptiveHash() ||
!estimatedOwnershipCount.isPresent()
+ ? clusterWriterCapacity
+ : Math.min(clusterWriterCapacity,
estimatedOwnershipCount.getAsLong());
+ }
+ String fallbackReason = distribution.getFallbackReason().orElse(null);
+ return new ExternalWriterParallelism((int) Math.max(1, planned),
+ estimatedOwnershipCount.isPresent() ?
estimatedOwnershipCount.getAsLong() : null,
+ fallbackReason);
+ }
+
+ private static OptionalLong estimateOwnershipCount(
+ ExternalWriteDistributionPlan distribution, Statistics statistics)
{
+ OptionalLong routingCap = estimateRoutingCardinalityCap(distribution);
+ if (statistics == null || !Double.isFinite(statistics.getRowCount())
+ || statistics.getRowCount() < 0) {
+ return routingCap;
+ }
+ if (statistics.getRowCount() == 0) {
+ return OptionalLong.of(1);
+ }
+ long rowCount = saturatedCeil(statistics.getRowCount());
+ long ownershipCount = 1;
+ for (NamedExpression route : distribution.getRoutingExpressions()) {
+ long routeCount = estimateExpressionCardinality(route, statistics,
rowCount);
+ long routeCap =
distribution.getRoutingCardinalityCap(route.getExprId())
+ .orElse(Long.MAX_VALUE);
+ ownershipCount = saturatedMultiply(
+ ownershipCount, Math.min(routeCount, routeCap), rowCount);
+ }
+ return OptionalLong.of(Math.max(1, Math.min(rowCount,
ownershipCount)));
+ }
+
+ private static OptionalLong estimateRoutingCardinalityCap(
+ ExternalWriteDistributionPlan distribution) {
+ long cap = 1;
+ for (NamedExpression route : distribution.getRoutingExpressions()) {
+ OptionalLong routeCap =
distribution.getRoutingCardinalityCap(route.getExprId());
+ if (!routeCap.isPresent()) {
+ return OptionalLong.empty();
+ }
+ cap = saturatedMultiply(cap, routeCap.getAsLong(), Long.MAX_VALUE);
+ }
+ return OptionalLong.of(Math.max(1, cap));
+ }
+
+ private static long estimateExpressionCardinality(
+ Expression expression, Statistics statistics, long rowCount) {
+ long cardinality = 1;
+ boolean foundKnownInput = false;
+ for (Slot input : expression.getInputSlots()) {
+ ColumnStatistic columnStatistic =
statistics.findColumnStatistics(input);
+ if (columnStatistic == null || columnStatistic.isUnKnown
Review Comment:
[P2] Keep capacity when any route input is unknown
For a composite routing expression, this skips unknown input statistics but
returns the product of the remaining known NDVs as soon as one is available.
Thus a fixed-bucket route with an NDV-1 key plus an unknown high-cardinality
key is estimated as one ownership unit and capped to one writer, even though
the unknown key can fill every bucket. That silently defeats this PR's scaling
goal on common incomplete statistics. Please treat the expression cardinality
as unknown when any relevant input is unknown (then retain the connector
cap/capacity), and add a mixed-known/unknown test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
