morningman commented on code in PR #68713:
URL: https://github.com/apache/doris/pull/68713#discussion_r4308625292
##########
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:
Fixed in 2c4665b82da. Confirmed, and it was a regression from the commit
that made each file declared by its own schema version (89ced4b0306): planning
the `fallbackOrcUsesEachBranchOptionsAndSchemas` fixture with the variable on
throws `RuntimeException: .../branch_0/schema/schema-2 (No such file or
directory)`.
Each split is now estimated by the branch that planned it.
`legacyOrcSplitTable` picks the branch and is renamed `splitBranch`, since both
paths use it now. Each branch gets an estimate of its own, which reads that
branch's schema directory. The helper also covers one more case: a split that
the pair's own snapshot reader plans (the file-creation OPTIONS path) is a
plain `DataSplit` of the wrapped branch, not a `FallbackSplit`, and the helper
used to cast it.
`PaimonScanPlanProviderTest.fallbackBranchJniSplitsDeclareByTheirOwnBranchSchemas`
reads both branches of such a table through JNI with the variable on. The
fallback branch's file is under schema 2, which the primary branch never had,
and has 1-byte row groups. Each split's declaration must match its own branch's
layout.
##########
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:
Fixed in 2c4665b82da.
- **Primary-key table:** fixed as you describe. Its diff split now declares
both file sets as one merge, since `IncrementalDiffSplitRead` keeps both merge
readers open.
- **Append table:** the "bypasses the gate" part needs a narrower scope
here. Its diff split goes to `RawFileSplitRead`, which reads only the
after-files, one after another, and ignores the before-files. So it now
declares the largest after-file's share.
- **Fallback-branch table:** its `IncrementalSplit` arrives wrapped in a
`FallbackSplitImpl` and is unwrapped first.
- **Other splits:** those that are not `DataSplit`s (metadata system tables)
still declare nothing.
Tests:
- `PaimonScanPlanProviderTest.incrementalDiffSplitDeclaresBothFileSets`
plans `incremental-between` with `diff` on a primary-key table. A full
compaction between the two snapshots makes both file sets non-empty, and the
test checks the declaration.
- `PaimonJniHeapEstimateTest` covers the append case and the wrapped one.
##########
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:
Fixed in 2c4665b82da. A parquet file's columns now get the
`parquet.dictionary.page.size` of the schema version the file was written
under, or 1 MB when that version does not set it. ORC files keep 1 MB a column,
since the option does not apply to them.
Two corrections to the arithmetic:
- For a primary-key table the allowance also counts the key, sequence and
row-kind columns.
- A decoded string dictionary holds a `Binary` per value on top of its page
bytes, so this term stays part of an estimate rather than a bound.
`PaimonJniHeapEstimateTest.parquetDictionaryPagesAreAsLargeAsTheSchemaVersionAllows`
covers it.
--
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]
