github-actions[bot] commented on code in PR #67726:
URL: https://github.com/apache/doris/pull/67726#discussion_r3969115593
##########
fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/CacheSpec.java:
##########
@@ -168,10 +196,115 @@ public static void checkLongProperty(String value, long
minValue, String key) {
}
}
+ /** Strict CREATE/ALTER-time validation for the catalog weight property. */
+ public static OptionalLong checkCatalogWeightProperty(Map<String, String>
properties) {
+ if (properties == null) {
+ return OptionalLong.empty();
+ }
+ String catalogValue =
properties.get(MetaCacheBudgetManager.CATALOG_MAX_WEIGHT_PROPERTY);
+ if (catalogValue == null) {
+ return OptionalLong.empty();
+ }
+ long parsed = parseWeight(catalogValue,
+ MetaCacheBudgetManager.CATALOG_MAX_WEIGHT_PROPERTY, false, 0L);
+ if (parsed <= 0L) {
+ throw new IllegalArgumentException(
+ MetaCacheBudgetManager.CATALOG_MAX_WEIGHT_PROPERTY + "
must be positive");
+ }
+ return OptionalLong.of(parsed);
+ }
+
+ /** Strict CREATE/ALTER-time validation for catalog and known engine entry
weight properties. */
+ public static void checkWeightProperties(
+ Map<String, String> properties, String engine, String...
knownEntries) {
+ OptionalLong catalogMax = checkCatalogWeightProperty(properties);
+ if (properties == null) {
+ return;
+ }
+ Set<String> entries = new HashSet<>();
+ Collections.addAll(entries, knownEntries);
+ String prefix = metaCacheKeyPrefix(engine);
+ for (Map.Entry<String, String> property : properties.entrySet()) {
+ String key = property.getKey();
+ if (!key.startsWith(prefix) || !key.endsWith(KEY_MAX_WEIGHT)) {
+ continue;
+ }
+ String entry = key.substring(prefix.length(), key.length() -
KEY_MAX_WEIGHT.length());
+ if (!entries.contains(entry)) {
+ throw new IllegalArgumentException("Unknown metadata cache
weight property: " + key);
Review Comment:
[P1] Preserve ALTER after downgrade. This binary can replay a catalog
containing a future `meta.cache.<engine>.<entry>.max-weight` key, but the
default update validator merges that key into every unrelated ALTER and this
branch rejects it; ALTER CATALOG cannot remove properties. Distinguish unknown
keys introduced by this statement from unknown keys already persisted so
CREATE/new typos may still fail, and add a future-key plus unrelated-ALTER test.
##########
fe/fe-connector/fe-connector-cache/src/main/java/org/apache/doris/connector/cache/ScopedMetaCache.java:
##########
@@ -366,10 +469,14 @@ void invalidateKey(
return;
}
afterStateReplacement.run();
- VersionedValue<K, V> registered = invalidated.node.registration.get();
- if (registered != null && registered.keyState == invalidated.keyState)
{
- data.asMap().remove(key, registered);
- invalidated.node.registration.compareAndSet(registered, null);
+ Registration registered = invalidated.node.registration.get();
+ VersionedValue<K, V> current = data.getIfPresent(key);
+ if (registered != null && current != null && current.registration ==
registered
+ && current.keyState == invalidated.keyState) {
+ data.asMap().remove(key, current);
+ }
+ if (registered != null &&
invalidated.node.registration.compareAndSet(registered, null)) {
Review Comment:
[P1] Keep invalidation cleanup tied to the invalidated generation. A put can
reuse this `KeyNode` after the state swap and install a new registration before
cleanup reaches this CAS; cleanup then leaves the new value in Caffeine but
clears/releases its reservation. Please compare against a registration tied to
the invalidated `KeyState` and add a deterministic same-node test without the
intervening `getIfPresent`.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergTableCache.java:
##########
@@ -130,7 +143,7 @@ TableLease borrow(TableIdentifier identifier,
Supplier<Table> loader) {
}
}
};
- TableOwner loaded = new TableOwner(table, cleanup, true);
+ TableOwner loaded = new TableOwner(table, cleanup, true,
entry.isWeightBounded());
Review Comment:
[P2] Skip weight preparation when this cache is disabled. With
`ttl-second=0` or `enable=false` plus a global/catalog limit, `isWeightBounded`
is still true, so every live table load serializes and estimates the full
`TableMetadata` even though the disabled path immediately discards it. Gate
this work on `entry.isEnabled()` as well and cover the combined configuration.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java:
##########
@@ -867,9 +878,27 @@ public void tryModifyCatalogProps(Map<String, String>
props) {
private void invalidateCachesAfterPropertyUpdate(Map<String, String>
updatedProps) {
ExternalMetaCacheMgr cacheMgr =
Env.getCurrentEnv().getExtMetaCacheMgr();
- if (updatedProps.get(SCHEMA_CACHE_TTL_SECOND) != null) {
+ if (updatedProps.get(SCHEMA_CACHE_TTL_SECOND) != null
+ ||
updatedProps.containsKey(MetaCacheBudgetManager.CATALOG_MAX_WEIGHT_PROPERTY)) {
Review Comment:
[P2] Retire old catalog budgets before allowing reinitialization. When this
ALTER changes the effective maximum, `resetToUninitialized` releases the
catalog monitor before `removeCatalog` runs, so a concurrent
`makeSureInitialized` can build the new connector while the default-engine
owner still holds the old catalog bucket; `createEntryBudget` then throws a
conflicting-limit error. Fence reinitialization through removal and add a real
old/new-budget interleaving test.
##########
fe/fe-connector/fe-connector-hive/src/main/java/org/apache/doris/connector/hive/HivePartitionViewSizeEstimator.java:
##########
@@ -0,0 +1,116 @@
+// 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.connector.hive;
+
+import org.apache.doris.connector.cache.ConnectorTableKey;
+import org.apache.doris.connector.cache.JvmSizeUtils;
+import org.apache.doris.connector.cache.MetaCacheSizeEstimate;
+import org.apache.doris.connector.cache.ReflectiveObjectSizeEstimator;
+import org.apache.doris.connector.spi.ConnectorPartitionInfo;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/** Bounded type-specific estimator for Hive's large derived partition view. */
+final class HivePartitionViewSizeEstimator {
+ private static final int PARTITION_SAMPLE_SIZE = 16;
+ private static final long KEY_SHALLOW_BYTES =
JvmSizeUtils.instanceSize(ConnectorTableKey.class);
+ private static final long PARTITION_SHALLOW_BYTES =
JvmSizeUtils.instanceSize(ConnectorPartitionInfo.class);
+ private static final long ARRAY_LIST_SHALLOW_BYTES =
JvmSizeUtils.instanceSize(ArrayList.class);
+ private static final long UNMODIFIABLE_LIST_SHALLOW_BYTES =
JvmSizeUtils.instanceSize(
+ Collections.unmodifiableList(Collections.emptyList()).getClass());
+ private static final long UNMODIFIABLE_MAP_SHALLOW_BYTES =
JvmSizeUtils.instanceSize(
+ Collections.unmodifiableMap(Collections.emptyMap()).getClass());
+ private static final long LINKED_HASH_MAP_SHALLOW_BYTES =
JvmSizeUtils.instanceSize(LinkedHashMap.class);
+ private static final long LINKED_HASH_MAP_ENTRY_SHALLOW_BYTES =
classSize("java.util.LinkedHashMap$Entry");
+
+ private HivePartitionViewSizeEstimator() {
+ }
+
+ static MetaCacheSizeEstimate estimateEntry(ConnectorTableKey key,
List<ConnectorPartitionInfo> partitions) {
+ long bytes = KEY_SHALLOW_BYTES;
+ bytes = add(bytes, JvmSizeUtils.stringSize(key.getDb()));
+ bytes = add(bytes, JvmSizeUtils.stringSize(key.getTable()));
+ bytes = add(bytes, ARRAY_LIST_SHALLOW_BYTES);
+ bytes = add(bytes, JvmSizeUtils.objectArraySize(partitions.size()));
+ bytes = add(bytes, JvmSizeUtils.sampledListPayload(
Review Comment:
[P1] Do not mark a sampled variable-length payload complete. A large
partition at an unsampled index is retained but absent from both this sample
and the reflective fallback, so the entry can be admitted below its real
weight. The same bug is present in both Iceberg derived-view estimators; use an
exhaustive construction-time payload pass (as Paimon does) or return
incomplete, and add skewed-tail tests.
##########
fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java:
##########
@@ -651,6 +651,10 @@ public class SchemaTable extends Table {
.column("LAST_LOAD_SUCCESS_TIME",
ScalarType.createStringType())
.column("LAST_LOAD_FAILURE_TIME",
ScalarType.createStringType())
.column("LAST_ERROR",
ScalarType.createStringType())
+ .column("MAX_WEIGHT",
ScalarType.createType(PrimitiveType.BIGINT))
Review Comment:
[P1] Keep the appended slots usable during a rolling upgrade. A new FE can
put `MAX_WEIGHT` (and the other new columns) in the destination tuple sent to
an old BE, but the old `SchemaScanOperatorX` rejects that unknown slot during
prepare, before this PR's FE-RPC 28-to-24 fallback can run. Gate
planning/routing by BE compatibility and add a mixed-version test that selects
and filters on `MAX_WEIGHT`.
--
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]
