924060929 commented on code in PR #66247:
URL: https://github.com/apache/doris/pull/66247#discussion_r3687961507
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonConnectorMetadata.java:
##########
@@ -1444,7 +1454,10 @@ public Optional<ConnectorTableStatistics>
getTableStatistics(
PaimonTableHandle paimonHandle = (PaimonTableHandle) handle;
long rowCount;
try {
- rowCount = catalogOps.rowCount(resolveTable(paimonHandle));
+ Table table = resolveTable(paimonHandle);
+ PaimonReaderOptions.validateEffectiveTable(table);
Review Comment:
[P2] Runtime-normalize manifest parallelism before statistics validation
Catalog/replay validation deliberately accepts hardware-independent values
up to 256, while execution copies cap them to `Runtime.availableProcessors()`.
This path validates the unresolved table directly. Thus, with local capacity
`N` and a catalog value `N+1 <= 256`, foreground scans are capped and succeed,
but row-count planning throws here, is caught below, and silently degrades to
`Optional.empty()`/UNKNOWN statistics (with a warning). The snapshot overload
has the same gap for non-OPTIONS pins. Please apply the shared runtime-safe
copy before validation and `rowCount`, and add an `availableProcessors()+1`
statistics case rather than only the zero-value rejection.
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -316,20 +317,60 @@ Table resolveTable(PaimonTableHandle paimonHandle) {
Table resolveScanTable(PaimonTableHandle paimonHandle) {
Table table = resolveTable(paimonHandle);
Map<String, String> scanOptions = paimonHandle.getScanOptions();
+ Table finalTable = table;
if (scanOptions != null && !scanOptions.isEmpty()) {
if (PaimonScanParams.isOptionsPin(scanOptions)) {
// An @options pin owns the whole scan-startup state:
applyOptions strips the internal
// markers and nulls out the absent members of paimon's
inherited read-state family, so a
// scan.mode / tag persisted on the base table cannot leak
into this relation's read.
- return PaimonScanParams.applyOptions(table, scanOptions);
+ finalTable = PaimonScanParams.applyOptions(table, scanOptions);
+ } else {
+ // FIX-INCR-SCAN-RESET: for an @incr read, reapply legacy's
null reset of
+ // scan.snapshot-id/scan.mode here (the single Table.copy
chokepoint shared by both the
+ // native/JNI scan path and the JNI serialized-table path) so
a stale persisted pin on the
+ // base table cannot hijack incremental-between.
Non-incremental pins pass through unchanged.
+ finalTable =
table.copy(PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions));
+ }
+ }
+ finalTable = runtimeSafeTable(finalTable);
+ // This is the last common boundary before planning and serialization.
Normalize and
+ // validate only after relation > catalog > physical precedence is
established.
+ PaimonReaderOptions.validateEffectiveTable(finalTable);
+ validateHiddenSystemDataTable(paimonHandle, scanOptions);
+ return finalTable;
+ }
+
+ private Table runtimeSafeTable(Table table) {
+ Map<String, String> runtimeOptions =
PaimonReaderOptions.runtimeSafeCopyOptions(
+ table, Collections.emptyMap());
+ // The cached catalog handle remains hardware-neutral; only the
query-local planning copy
+ // receives a CPU-local cap before it can resize Paimon's JVM-wide
manifest executor.
+ return runtimeOptions.isEmpty() ? table : table.copy(runtimeOptions);
+ }
+
+ private void validateHiddenSystemDataTable(PaimonTableHandle handle,
Map<String, String> scanOptions) {
+ if (!handle.isSystemTable()) {
+ return;
+ }
+ try {
+ Table dataTable = handle.getSystemTableSource();
+ if (dataTable == null) {
+ dataTable = handle.getSysBaseTable();
+ }
+ if (dataTable == null) {
+ dataTable = catalogOps.getTable(
+ Identifier.create(handle.getDatabaseName(),
handle.getTableName()));
}
- // FIX-INCR-SCAN-RESET: for an @incr read, reapply legacy's null
reset of
- // scan.snapshot-id/scan.mode here (the single Table.copy
chokepoint shared by both the
- // native/JNI scan path and the JNI serialized-table path) so a
stale persisted pin on the
- // base table cannot hijack incremental-between. Non-incremental
pins pass through unchanged.
- return
table.copy(PaimonIncrementalScanParams.applyResetsIfIncremental(scanOptions));
+ if (PaimonScanParams.isOptionsPin(scanOptions)) {
+ // Read-only system wrappers plan manifests through their
hidden data table, so the
+ // same relation copy must establish precedence on both
visible and hidden handles.
+ PaimonScanParams.applyOptions(dataTable, scanOptions);
Review Comment:
[P1] Apply the runtime cap to the system wrapper's actual hidden table
For Paimon read-only system wrappers, the outer `Table.options()` is empty
while planning runs through the private `FileStoreTable`. On an FE with `N <
256` processors, take a physical/catalog `scan.manifest.parallelism=N+1` and
query `t$partitions@options('read.batch-size'='4096')`. `applyOptions` on the
wrapper copies only the reader option, so its hidden table still inherits
`N+1`; `runtimeSafeTable` and `validateEffectiveTable` then see the wrapper's
empty option map. This call does cap and validate a separate `dataTable` copy,
but its return value is discarded, and the wrapper returned by
`resolveScanTable` still plans with `N+1`, allowing Paimon to enlarge its
JVM-static manifest executor. Please normalize the exact hidden source used by
the returned wrapper (for example, rebuild the wrapper from the effective
source), and cover an unrelated relation option combined with an
above-local-CPU manifest value.
--
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]
