github-actions[bot] commented on code in PR #68713:
URL: https://github.com/apache/doris/pull/68713#discussion_r4308501021
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -948,7 +962,7 @@ && hasTrustworthyLimitAccounting(table)
validateMetadataColumnReader(true, false);
}
ranges.add(buildJniScanRange(split, defaultFileFormat,
- Collections.emptyMap(), false, weightDenominator));
+ Collections.emptyMap(), false, weightDenominator, 0));
Review Comment:
[P2] Declare heap for incremental Paimon merge splits
The supported `@incr('startSnapshotId'=1, 'endSnapshotId'=2,
'incrementalBetweenScanMode'='diff')` path returns `IncrementalSplit`, which is
not a `DataSplit` and therefore receives this literal zero even with admission
enabled. Paimon 1.4.2's
[IncrementalDiffSplitRead](https://github.com/apache/paimon/blob/release-1.4.2/paimon-core/src/main/java/org/apache/paimon/table/source/splitread/IncrementalDiffSplitRead.java#L81)
creates merge readers over both `beforeFiles` and `afterFiles` and retains
them together. Large diff reads consequently bypass the gate completely and can
still exhaust the JVM when several scanners run, unlike ordinary merge splits.
Estimate these file sets too, including wrapped incremental splits, and test
that the diff scan sends a positive declaration. This is separate from the
existing issue about decoded peaks above a positive declaration.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonJniHeapEstimate.java:
##########
@@ -0,0 +1,155 @@
+// 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.paimon;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.MemorySize;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.DataSplit;
+
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.function.LongFunction;
+
+/**
+ * The JVM heap BE's JNI reader of a paimon {@link DataSplit} holds, declared
to BE's JNI heap gate
+ * ({@code TFileRangeDesc.jni_heap_bytes}) for a statement that sets {@code
enable_jni_heap_admission}.
+ *
+ * <p>A JNI read of a split whose files overlap - a primary-key table written
to since it was last
+ * compacted - merges them: paimon opens one file of every sorted run of a
section at once, and an open
+ * parquet file keeps the compressed column chunks of its current row group in
the heap until it moves
+ * on to the next, when the old and the new are there together for a moment.
ORC keeps a stripe the
+ * same way. Sixteen such splits read at once are what runs a 2 GB heap out.
So a split that has to be
+ * merged holds, steadily, the sum over its files of one row group - two for a
file that has more than one
+ * - plus a dictionary page for every column of every file. That takes all the
files as one section,
+ * which is exact for the uncompacted tables that need the gate and over the
mark for files that do not
+ * overlap, which are read one section after another. A split that needs no
merging reads its files one
+ * after another and holds one file's share.
+ *
+ * <p>A file's row groups and columns are those of the schema version it was
written under
+ * ({@link DataFileMeta#schemaId()}): paimon's writers take the row group size
from the table's options,
+ * and an ALTER TABLE that changes them, or the columns, leaves the files
written before it as they are.
+ * A write job can also pass the row group size as a dynamic option, which no
schema version records; a
+ * file written so is taken by its schema version's options, and declared
below what it holds when the
+ * job's row groups were larger.
+ *
+ * <p>That steady hold is what is declared, not everything a merge ever holds:
while it decodes, its column
+ * batches add brief peaks on top, measured at about three times the
declaration for a split of a 126 MB
+ * file read on its own. The peaks are left to the heap outside the budget -
the other half of it with
+ * jni_scanner_heap_budget_ratio at its default - which has room for some of
the admitted readers to peak
+ * at once but not for all of them; a read that needs more fails as it does
with enable_jni_heap_admission
+ * off.
+ *
+ * <p>Every column is taken as read. A narrow projection reads less, but a
merge also reads every key
+ * column, the sequence number and the row kind whatever is projected, and
declaring less than a merge
+ * holds steadily would let the gate admit more readers than its budget fits.
+ */
+final class PaimonJniHeapEstimate {
+
+ // What paimon's writers fall back to when neither file.block-size nor the
format's own option is
+ // set (CoreOptions.FILE_BLOCK_SIZE).
+ static final long DEFAULT_PARQUET_ROW_GROUP_BYTES = 128L * 1024 * 1024;
+ static final long DEFAULT_ORC_STRIPE_BYTES = 64L * 1024 * 1024;
+ // The largest dictionary page paimon's parquet writer keeps for a column.
+ static final long DICTIONARY_BYTES_PER_COLUMN = 1024L * 1024;
+
+ // How the files written under a schema version are laid out, by its id,
each worked out once.
+ private final LongFunction<FileLayout> layoutOfSchema;
+ private final Map<Long, FileLayout> layouts = new HashMap<>();
+
+ PaimonJniHeapEstimate(LongFunction<FileLayout> layoutOfSchema) {
+ this.layoutOfSchema = layoutOfSchema;
+ }
+
+ /**
+ * For the splits of {@code table}, each file laid out by the schema
version it was written under:
+ * the table's own is at hand, and {@code schemaAt} reads the others by id.
+ */
+ static PaimonJniHeapEstimate of(FileStoreTable table,
LongFunction<TableSchema> schemaAt) {
+ TableSchema own = table.schema();
+ return new PaimonJniHeapEstimate(
+ schemaId -> FileLayout.of(schemaId == own.id() ? own :
schemaAt.apply(schemaId)));
+ }
+
+ /** For a table whose schema versions are not at hand: every file laid out
by the options it shows. */
+ static PaimonJniHeapEstimate of(Table table) {
+ FileLayout layout = FileLayout.of(table.options(),
table.rowType().getFieldCount(),
+ table.primaryKeys().size());
+ return new PaimonJniHeapEstimate(schemaId -> layout);
+ }
+
+ long bytesOf(DataSplit split) {
+ long total = 0;
+ long largest = 0;
+ for (DataFileMeta file : split.dataFiles()) {
+ long share = layouts.computeIfAbsent(file.schemaId(),
layoutOfSchema::apply).bytesHeld(file);
+ total += share;
+ largest = Math.max(largest, share);
+ }
+ return split.rawConvertible() ? largest : total;
+ }
+
+ /** The row group size and the columns of the files written under one
schema version. */
+ static final class FileLayout {
+ private final long parquetRowGroupBytes;
+ private final long orcStripeBytes;
+ private final long dictionaryBytes;
+
+ FileLayout(long parquetRowGroupBytes, long orcStripeBytes, int
columns) {
+ this.parquetRowGroupBytes = parquetRowGroupBytes;
+ this.orcStripeBytes = orcStripeBytes;
+ this.dictionaryBytes = columns * DICTIONARY_BYTES_PER_COLUMN;
Review Comment:
[P2] Honor the configured Parquet dictionary-page size
The 1 MiB allowance is only the writer's default. Paimon 1.4.2 forwards the
stored `parquet.dictionary.page.size` option to
[withDictionaryPageSize](https://github.com/apache/paimon/blob/release-1.4.2/paimon-format/src/main/java/org/apache/paimon/format/parquet/writer/RowDataParquetBuilder.java#L77),
and its reader retains the initialized, decompressed dictionaries throughout
the row group. With a 16 MiB limit, eight string columns can retain 8 MiB
dictionaries each even when common prefixes compress the physical file below 8
MiB. `bytesHeld` then charges the small `fileSize` plus only 8 MiB for those
dictionaries, although the dictionaries alone retain 64 MiB; the nominal
row-group size supplies no cushion on this branch. Derive this allowance from
each file's historical schema options. This is steady retained memory, separate
from the already-reported transient decoded-batch peaks.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -2009,6 +2026,37 @@ && containsVariant(field.type()))) {
return physicalVariantSchemaIds;
}
+ /**
+ * What the JNI DataSplits of {@code table} declare
(PaimonJniHeapEstimate). A file holds row groups of
+ * the size set by the schema version it was written under, which an ALTER
TABLE since leaves as it is,
+ * so the estimate reads the versions its files were written under. $ro
reads its base table's files,
+ * under that table's schema ids, as {@link #physicalVariantSchemaIds}
resolves it; the other system
+ * tables are estimated by the options they show.
+ */
+ private PaimonJniHeapEstimate jniHeapEstimate(Table table,
PaimonTableHandle handle) {
+ Table physicalSchemaTable = resolveSchemaDictTable(table, handle);
+ if (!(physicalSchemaTable instanceof FileStoreTable)) {
+ return PaimonJniHeapEstimate.of(table);
+ }
+ FileStoreTable fileStoreTable = (FileStoreTable) physicalSchemaTable;
+ SchemaManager schemaManager = fileStoreTable.schemaManager();
Review Comment:
[P2] Resolve each split's branch before reading its schema
The single estimator takes `schema()` and `schemaManager()` from
`FallbackReadFileStoreTable`, which delegates both to its wrapped branch, but
its `FallbackDataSplit`s can contain files from the other branch. Schema IDs
are branch-local: the existing `fallbackOrcUsesEachBranchOptionsAndSchemas`
fixture has wrapped schema 0 and fallback schema 2, with the fallback ORC split
routed to JNI. Enabling `enable_jni_heap_admission` makes this code look up
schema 2 in the wrapped branch and fail an otherwise valid query during
planning; colliding IDs can instead use the wrong layout. Select the physical
branch as `legacyOrcSplitTable` already does and keep estimators/caches per
branch. This is distinct from the earlier within-table historical-option
finding: the new lookup uses the wrong schema namespace even without changing
block-size options.
--
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]
