github-actions[bot] commented on code in PR #66247:
URL: https://github.com/apache/doris/pull/66247#discussion_r3690865844
##########
fe/fe-connector/fe-connector-paimon/src/main/java/org/apache/doris/connector/paimon/PaimonScanPlanProvider.java:
##########
@@ -316,20 +322,48 @@ 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 = PaimonReaderOptions.runtimeSafeTable(finalTable);
+ finalTable = runtimeSafeSystemTable(paimonHandle, finalTable,
scanOptions);
+ // This is the last common boundary before planning and serialization.
Normalize and
+ // validate only after relation > catalog > physical precedence is
established.
+ PaimonReaderOptions.validateEffectiveTable(finalTable);
+ return finalTable;
+ }
+
+ private Table runtimeSafeSystemTable(
+ PaimonTableHandle handle, Table systemTable, Map<String, String>
scanOptions) {
+ if (!handle.isSystemTable()) {
+ return systemTable;
+ }
+ try {
+ Table dataTable = handle.getSystemTableSource();
+ if (dataTable == null) {
+ dataTable = handle.getSysBaseTable();
+ }
+ if (dataTable == null) {
+ dataTable = catalogOps.getTable(
Review Comment:
[P1] Authenticate the fallback system-source reload
For a deserialized/plan-reused system handle, `resolveTable` correctly
reloads the system wrapper inside `executeAuthenticated`, but all three
transient source fields are still null and this newly added second catalog
lookup runs directly. On a Kerberos-backed catalog that base-table RPC
therefore executes outside the injected UGI/TCCL and can fail even though the
wrapper reload succeeded; the metadata statistics twin has the same direct
lookup. Please route both fallbacks through the connector authentication
context (or a shared authenticated base-reload helper) and cover a
transient-free system handle.
##########
fe/be-java-extensions/paimon-scanner/src/main/java/org/apache/doris/paimon/PaimonJniScanner.java:
##########
@@ -523,64 +627,146 @@ static Optional<Long> parseDataSizeBytes(String value) {
if (value == null || value.trim().isEmpty()) {
return Optional.empty();
}
- String normalized = value.trim().toLowerCase(Locale.ROOT).replace("_",
"").replace(" ", "");
- int unitStart = 0;
- while (unitStart < normalized.length()
- && (Character.isDigit(normalized.charAt(unitStart)) ||
normalized.charAt(unitStart) == '.')) {
- unitStart++;
- }
- if (unitStart == 0) {
- return Optional.empty();
- }
try {
- double number = Double.parseDouble(normalized.substring(0,
unitStart));
- String unit = normalized.substring(unitStart);
- long multiplier;
- switch (unit) {
- case "":
- case "b":
- case "byte":
- case "bytes":
- multiplier = 1L;
- break;
- case "k":
- case "kb":
- case "kib":
- multiplier = 1024L;
- break;
- case "m":
- case "mb":
- case "mib":
- multiplier = 1024L * 1024L;
- break;
- case "g":
- case "gb":
- case "gib":
- multiplier = 1024L * 1024L * 1024L;
- break;
- case "t":
- case "tb":
- case "tib":
- multiplier = 1024L * 1024L * 1024L * 1024L;
- break;
- default:
- return Optional.empty();
- }
- return Optional.of((long) (number * multiplier));
- } catch (NumberFormatException e) {
+ // Keep the BE guard's accepted grammar identical to the Paimon
option parser that will
+ // consume this value; accepting a superset lets invalid
serialized options reach scans.
+ return Optional.of(MemorySize.parse(value).getBytes());
+ } catch (IllegalArgumentException e) {
return Optional.empty();
}
}
private void initTable() {
Preconditions.checkState(params.containsKey("serialized_table"));
table = PaimonUtils.deserialize(params.get("serialized_table"));
+ String encodedSystemSource = params.get(PAIMON_OPTION_PREFIX +
DORIS_SERIALIZED_SYSTEM_SOURCE);
+ FileStoreTable systemSource = encodedSystemSource == null
+ ? null : PaimonUtils.deserialize(encodedSystemSource);
+ table = applyBackendManifestParallelism(table,
+ params.get(PAIMON_OPTION_PREFIX +
DORIS_MANIFEST_PARALLELISM_CAP),
+ Runtime.getRuntime().availableProcessors(), systemSource,
+ params.get(PAIMON_OPTION_PREFIX + DORIS_SYSTEM_TABLE_TYPE));
+ validateSerializedReaderOptions(table);
paimonAllFieldNames = PaimonUtils.getFieldNames(this.table.rowType());
if (LOG.isDebugEnabled()) {
LOG.debug("paimonAllFieldNames:{}", paimonAllFieldNames);
}
}
+ static Table applyBackendManifestParallelism(
+ Table table, String feParallelismCap, int localCapacity) {
+ return applyBackendManifestParallelism(
+ table, feParallelismCap, localCapacity, null, null);
+ }
+
+ static Table applyBackendManifestParallelism(
+ Table table, String feParallelismCap, int localCapacity,
+ FileStoreTable systemSource, String systemTableType) {
+ Table planningTable = systemSource == null ? table : systemSource;
+ List<Integer> configuredValues = new ArrayList<>();
+ collectManifestParallelism(planningTable, configuredValues);
+ int requestedBound = localCapacity;
Review Comment:
[P1] Keep the stable 256 cap in the old-FE backstop
When this BE is paired with an older FE, `feParallelismCap` is absent, so
`requestedBound` is only `localCapacity`. On a host exposing more than 256
processors, an ordinary or fallback serialized planner with
`scan.manifest.parallelism=300` is therefore returned unchanged (and 600 is
only lowered to the host count), even though the connector deliberately defines
256 as the hardware-independent ceiling for Paimon's JVM-global manifest
executor. Please include that stable maximum in this backstop and cover a null
FE cap with `localCapacity > 256`, including a hidden fallback child.
--
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]
