github-actions[bot] commented on code in PR #68885:
URL: https://github.com/apache/doris/pull/68885#discussion_r4237454005
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java:
##########
@@ -767,57 +785,67 @@ static Table applyBackendManifestParallelism(
private static Table applyManifestParallelismBound(
Table table, int safeBound, boolean materializeAbsent) {
+ return adjustReaders(table, reader -> boundManifestParallelism(reader,
safeBound, materializeAbsent));
+ }
+
+ private static FileStoreTable boundManifestParallelism(
+ FileStoreTable table, int safeBound, boolean materializeAbsent) {
+ String configured =
table.options().get(CoreOptions.SCAN_MANIFEST_PARALLELISM.key());
+ if (configured == null && !materializeAbsent) {
+ return table;
+ }
+ if (configured != null && parsePositiveManifestParallelism(configured)
<= safeBound) {
+ return table;
+ }
+ // An absent option inherits Paimon's processor-count default, so
materialize the stable
+ // bound instead of assuming that an empty option map is already safe
on large hosts.
+ Map<String, String> cap = Collections.singletonMap(
+ CoreOptions.SCAN_MANIFEST_PARALLELISM.key(),
String.valueOf(safeBound));
+ return table.copyWithoutTimeTravel(cap);
+ }
+
+ /**
+ * Applies {@code adjust} to every table that reads for {@code table}:
each branch of a fallback
+ * pair on its own, the table behind a delegate, and the source a
read-only system wrapper hides.
+ * The wrappers around a changed table are rebuilt; an unchanged table is
returned as is.
+ */
+ private static Table adjustReaders(Table table,
UnaryOperator<FileStoreTable> adjust) {
if (table instanceof FallbackReadFileStoreTable) {
FallbackReadFileStoreTable pair = (FallbackReadFileStoreTable)
table;
- FileStoreTable main = applyManifestParallelismBound(
- pair.wrapped(), safeBound, materializeAbsent);
- FileStoreTable other = applyManifestParallelismBound(
- pair.other(), safeBound, materializeAbsent);
+ FileStoreTable main = adjustReaders(pair.wrapped(), adjust);
+ FileStoreTable other = adjustReaders(pair.other(), adjust);
Review Comment:
[P1] Preserve the chain fallback subtype when adjusting readers. In Paimon
1.4.2, `FileStoreTableFactory.createChainTable` nests a `ChainGroupReadTable`
(itself a fallback subclass) as the outer pair's other branch. With no
`read.batch-size`, this recursion rebuilds the inner table as a plain
`FallbackReadFileStoreTable`; the outer constructor rejects any nested fallback
other than `ChainGroupReadTable`, so a plain
`rest_catalog.db.chain_table$files` JNI scan fails in `initTable` before
reading. That system-table route skips the ordinary MVCC pin and can carry the
chain shape to BE. Preserve the specialized subtype and cover this scan.
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java:
##########
@@ -725,13 +726,30 @@ private void initTable() {
params.get(PAIMON_OPTION_PREFIX +
DORIS_MANIFEST_PARALLELISM_CAP),
Runtime.getRuntime().availableProcessors(), systemSource,
params.get(PAIMON_OPTION_PREFIX + DORIS_SYSTEM_TABLE_TYPE));
- validateSerializedReaderOptions(table);
+ table = applyDefaultReadBatchSize(table, batchSize);
paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType());
if (LOG.isDebugEnabled()) {
LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames);
}
}
+ /**
+ * Validates the serialized reader options and gives every reader of the
table that has no
+ * {@code read.batch-size} (from the table, the catalog or the relation)
the Doris batch size
+ * instead of Paimon's default. A configured value wins: Doris' output
block size and Paimon's
+ * reader batch are independent controls.
+ */
+ static Table applyDefaultReadBatchSize(Table table, int dorisBatchSize) {
+ validateSerializedReaderOptions(table);
+ Map<String, String> readOptions = Collections.singletonMap(
+ CoreOptions.READ_BATCH_SIZE.key(),
String.valueOf(dorisBatchSize));
+ // The serialized table may pin an older data snapshot while carrying
the latest schema
+ // after a schema change. A normal copy would time travel to that
snapshot's schema again
+ // and make renamed or newly added columns disappear.
+ return adjustReaders(table, reader ->
reader.options().containsKey(CoreOptions.READ_BATCH_SIZE.key())
Review Comment:
[P2] Bound the physical Paimon batch for legacy adaptive JNI scans.
`FileScanner` opens the V1 Java reader with the stable session `batch_size`
(8160 by default) before its first adaptive 32-row output request. This new
default fixes Paimon's `TableRead` at 8160; later `set_batch_size` only changes
the Java output block, while the physical iterator remains live across blocks.
Paimon ORC/Parquet allocate read vectors from that physical batch size, so
wide-row or short-LIMIT scans can allocate and decode about eight times the
prior 1024-row batch while Doris requests 32 rows. Bound the physical batch or
apply adaptive updates at a safe reader boundary, and test the physical read
path.
--
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]
