yujun777 commented on code in PR #69850: URL: https://github.com/apache/doris/pull/69850#discussion_r4424597580
########## fe/fe-core/src/main/java/org/apache/doris/catalog/stream/PartitionStreamState.java: ########## @@ -0,0 +1,131 @@ +// 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.catalog.stream; + +import com.google.gson.annotations.SerializedName; + +/** + * What a table stream recorded for one partition when it last advanced that partition's offset. + * + * <p>A changelog read returns the change rows between the stream offset and the end TSO of the + * read, a number the FE cannot derive from table metadata: the changelog index of a partition + * holds the change rows of every offset range, so its row count is the size of the whole + * changelog, not of the range being read. Two counters recorded next to the offset give the + * estimate a baseline to subtract: + * + * <ul> + * <li>{@code changelogRowCount} (P): the row count of the partition's changelog index at the + * moment the offset advanced; the read estimate is the current row count minus it;</li> + * <li>{@code baseRowCount} (Q): the row count of the partition's base index at the same moment; + * a snapshot read answers with the image at the offset, and Q is its size.</li> + * </ul> + * + * <p>The rule behind the estimates is to err on the large side: over-estimating a relation only + * makes the optimizer plan it differently, while under-estimating lets a large relation become + * the build side of a broadcast join and exhaust memory. + */ +public class PartitionStreamState { + public static final long UNKNOWN_ROW_COUNT = -1L; + + @SerializedName("crc") + private final long changelogRowCount; + + @SerializedName("brc") + private final long baseRowCount; + + public PartitionStreamState(long changelogRowCount, long baseRowCount) { + this.changelogRowCount = changelogRowCount; + this.baseRowCount = baseRowCount; + } + + public long getChangelogRowCount() { + return changelogRowCount; + } + + public long getBaseRowCount() { + return baseRowCount; + } + + /** + * Change rows the read of {@code (startTso, endTso]} may return, or {@link #UNKNOWN_ROW_COUNT} + * when no counter can tell. + * + * <p>A read of an empty range returns nothing no matter what the row counts report: the two + * TSOs are commit metadata that do not lag behind, unlike the row counts, so they decide the + * empty case. Otherwise the rows still in the changelog bound the read: a valid read only + * returns rows which were not reclaimed yet. + */ + public static long changelogDelta(long startTso, long endTso, long storedChangelogRowCount, + long currentChangelogRowCount) { + if (startTso < 0 || endTso < 0) { + // The range of the read is not known, so the difference of the counters cannot be + // attributed to it: the read may start before the counters do. + return UNKNOWN_ROW_COUNT; + } + if (endTso <= startTso) { + return 0; + } + if (storedChangelogRowCount >= 0 && currentChangelogRowCount > storedChangelogRowCount) { + return currentChangelogRowCount - storedChangelogRowCount; + } + if (currentChangelogRowCount > 0) { + // No baseline, or the counters moved backwards (a reclamation, or a partition which + // was recreated): every row the read may return is in the changelog right now. + return currentChangelogRowCount; + } + return UNKNOWN_ROW_COUNT; + } + + /** + * Estimate of a changelog read. See {@link #changelogDelta}. When the row counts cannot bound + * the read (not reported yet, or the partition was just written and the backend has not + * reported it), fall back to values which keep the estimate in the right order of magnitude + * instead of returning one row: a base row can be changed many times, so the base row count + * is not a strict bound, it only keeps the relation from looking tiny. + */ + public static long estimateChangelogRead(long startTso, long endTso, long storedChangelogRowCount, + long currentChangelogRowCount, long currentBaseRowCount, long tableRowCount) { + long delta = changelogDelta(startTso, endTso, storedChangelogRowCount, currentChangelogRowCount); + if (delta < 0) { + delta = Math.max(currentChangelogRowCount, currentBaseRowCount); + if (delta <= 0) { + delta = tableRowCount; + } + } + return Math.max(1, delta); Review Comment: Fixed in 6d2b9511c74. When the TSOs say the range holds changes and every counter is unavailable (the changelog count, the partition's base count and the table record are all zero), `estimateChangelogRead` no longer falls through to `Math.max(1, ...)`: it reports `UNKNOWN_NONEMPTY_ROW_COUNT` (`1L << 30`), a relation deliberately too large to be broadcast, because the only thing known there is that the range is not empty. An empty range still answers with one row, which the TSO comparison decides without the counters. `PartitionStreamStateTest` pins both cases (`testAKnownNonemptyRangeWithoutCountersIsNotEstimatedAtOneRow`), including the unreported-counts case that used to expect one row. The remaining P2 thread on the partition-pruned fallback is unchanged by design: with the estimates in place the changelog wrapper no longer reports `UNKNOWN_ROW_COUNT` for a stream read, so the fabrication it warns about only applies to a wrapper without a recorded estimate. -- 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]
