github-actions[bot] commented on code in PR #66473:
URL: https://github.com/apache/doris/pull/66473#discussion_r3801424905
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/paimon/source/PaimonScanNode.java:
##########
@@ -745,52 +750,243 @@ public List<org.apache.paimon.table.source.Split>
getPaimonSplitFromAPI() throws
if (PaimonScanParams.isPinnedEmptyScan(resolvedOptions)) {
return Collections.emptyList();
}
- Optional<Long> fileCreationTime =
PaimonScanParams.getPinnedFileCreationTime(resolvedOptions);
- if (fileCreationTime.isPresent()) {
- if (!(paimonTable instanceof FileStoreTable)) {
- throw new UserException("Paimon file-creation OPTIONS
require a data table.");
+ int[] projectedColumns = new int[0];
+ if
(!PaimonScanParams.getPinnedFileCreationTime(resolvedOptions).isPresent()) {
+ List<String> fieldNames =
paimonTable.rowType().getFieldNames();
+ projectedColumns = desc.getSlots().stream().mapToInt(
+ slot -> getFieldIndex(fieldNames,
slot.getColumn().getName()))
+ .toArray();
+ if (Arrays.stream(projectedColumns).anyMatch(index -> index <
0)) {
+ throw new UserException("Paimon scan schema does not
contain all bound Doris columns.");
}
- FileStoreTable fileStoreTable = (FileStoreTable) paimonTable;
- SnapshotReader snapshotReader =
fileStoreTable.newSnapshotReader()
- .withMode(ScanMode.ALL)
- .withSnapshot(Long.parseLong(
-
paimonTable.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key())))
- .withManifestEntryFilter(entry ->
- entry.file().creationTimeEpochMillis() >=
fileCreationTime.get());
- preserveBatchScanFilters(fileStoreTable, snapshotReader);
- if (predicates != null) {
- predicates.forEach(snapshotReader::withFilter);
- }
- return snapshotReader.read().splits();
- }
- List<String> fieldNames = paimonTable.rowType().getFieldNames();
- int[] projected = desc.getSlots().stream().mapToInt(
- slot -> getFieldIndex(fieldNames,
slot.getColumn().getName()))
- .toArray();
- if (Arrays.stream(projected).anyMatch(index -> index < 0)) {
- throw new UserException("Paimon scan schema does not contain
all bound Doris columns.");
}
- ReadBuilder readBuilder = paimonTable.newReadBuilder();
- TableScan scan = readBuilder.withFilter(predicates)
- .withProjection(projected)
- .newScan();
- PaimonMetricRegistry registry = new PaimonMetricRegistry();
- if (scan instanceof InnerTableScan) {
- scan = ((InnerTableScan) scan).withMetricRegistry(registry);
+ int[] projected = projectedColumns;
+ PaimonSplitTaskCacheKey cacheKey = createPaimonSplitTaskCacheKey(
+ relationSnapshot, paimonTable, resolvedOptions,
+ scanParams != null && scanParams.incrementalRead()
+ ? getIncrReadParams() : Collections.emptyMap(),
+ projected);
+ if (!canReuseExternalScanTasks()
+ && !(source.getExternalTable() instanceof
PaimonSysExternalTable)) {
+ // Reuse is off (or no statement cache): consume the native
splits directly, so an
+ // opt-out never pays the serialization cost of the cache
path. System tables keep
+ // the serialized path: their splits must be isolated copies
for the JNI reader.
+ return planPaimonSplits(paimonTable, resolvedOptions,
projected);
}
- List<org.apache.paimon.table.source.Split> splits =
scan.plan().splits();
- PaimonScanMetricsReporter.report(source.getTargetTable(),
paimonTable.name(), registry);
- if (!registry.getAllGroups().isEmpty()) {
- registry.clear();
+ List<PaimonSerializedScanTask> serializedSplits;
+ try {
+ serializedSplits = getOrLoadExternalScanTasks(cacheKey,
+ remainingBytes -> serializePaimonSplitsWithinLimit(
+ planPaimonSplits(paimonTable, resolvedOptions,
projected), remainingBytes),
+ PaimonScanNode::serializedTaskBytes,
+
StatementContext.ExternalScanTaskCache.WeightBudget.PAIMON_SERIALIZED_BYTES,
+ maxRetainedSerializedTaskBytes,
maxRetainedSerializedTaskBytes, true);
+ } catch (PaimonTaskCacheLimitException e) {
+ List<org.apache.paimon.table.source.Split> uncachedSplits =
e.takePlannedSplits();
+ return uncachedSplits == null
+ ? planPaimonSplits(paimonTable, resolvedOptions,
projected)
+ : uncachedSplits;
}
- return splits;
+ return serializedSplits.stream()
+ .map(PaimonSerializedScanTask::deserialize)
+ .collect(Collectors.toList());
+ } catch (UserException e) {
+ throw e;
+ } catch (Exception e) {
+ throw new UserException("Failed to plan Paimon scan tasks", e);
} finally {
if (getSummaryProfile() != null) {
getSummaryProfile().addExternalTableGetFileScanTasksTime(System.currentTimeMillis()
- startTime);
}
}
}
+ private List<org.apache.paimon.table.source.Split> planPaimonSplits(
+ Table paimonTable, Map<String, String> resolvedOptions, int[]
projected) throws UserException {
+ Optional<Long> fileCreationTime =
PaimonScanParams.getPinnedFileCreationTime(resolvedOptions);
+ if (fileCreationTime.isPresent()) {
+ if (!(paimonTable instanceof FileStoreTable)) {
+ throw new UserException("Paimon file-creation OPTIONS require
a data table.");
+ }
+ FileStoreTable fileStoreTable = (FileStoreTable) paimonTable;
+ SnapshotReader snapshotReader = fileStoreTable.newSnapshotReader()
+ .withMode(ScanMode.ALL)
+ .withSnapshot(Long.parseLong(
+
paimonTable.options().get(CoreOptions.SCAN_SNAPSHOT_ID.key())))
+ .withManifestEntryFilter(entry ->
+ entry.file().creationTimeEpochMillis() >=
fileCreationTime.get());
+ preserveBatchScanFilters(fileStoreTable, snapshotReader);
+ if (predicates != null) {
+ predicates.forEach(snapshotReader::withFilter);
+ }
+ return snapshotReader.read().splits();
+ }
+ ReadBuilder readBuilder = paimonTable.newReadBuilder();
+ TableScan scan = readBuilder.withFilter(predicates)
+ .withProjection(projected)
+ .newScan();
+ PaimonMetricRegistry registry = new PaimonMetricRegistry();
+ if (scan instanceof InnerTableScan) {
+ scan = ((InnerTableScan) scan).withMetricRegistry(registry);
+ }
+ List<org.apache.paimon.table.source.Split> splits =
scan.plan().splits();
+ PaimonScanMetricsReporter.report(source.getTargetTable(),
paimonTable.name(), registry);
+ if (!registry.getAllGroups().isEmpty()) {
+ registry.clear();
+ }
+ return splits;
+ }
+
+ private PaimonSplitTaskCacheKey createPaimonSplitTaskCacheKey(
+ Optional<MvccSnapshot> relationSnapshot, Table paimonTable,
+ Map<String, String> resolvedOptions, Map<String, String>
incrementalOptions,
+ int[] projected) {
+ Long snapshotId = null;
+ Long schemaId = null;
+ if (relationSnapshot.isPresent() && relationSnapshot.get() instanceof
PaimonMvccSnapshot) {
+ PaimonSnapshot snapshot = ((PaimonMvccSnapshot)
relationSnapshot.get())
+ .getSnapshotCacheValue().getSnapshot();
+ snapshotId = snapshot.getSnapshotId();
+ schemaId = snapshot.getSchemaId();
+ }
+ return new PaimonSplitTaskCacheKey(
+ source.getCatalog().getId(),
+ source.getExternalTable().getId(),
+ source.getTargetTable().getId(),
+ snapshotId,
+ schemaId,
+ scanParams == null ? null : scanParams.getParamType(),
+ scanParams == null ? Collections.emptyList() :
scanParams.getListParams(),
+ resolvedOptions,
+ incrementalOptions,
+ paimonTable.options(),
+ projected,
+ PaimonUtil.encodeObjectToString(predicates));
+ }
+
+ @VisibleForTesting
+ void setMaxRetainedSerializedTaskBytes(long
maxRetainedSerializedTaskBytes) {
+ this.maxRetainedSerializedTaskBytes = maxRetainedSerializedTaskBytes;
+ }
+
+ private List<PaimonSerializedScanTask> serializePaimonSplitsWithinLimit(
+ List<org.apache.paimon.table.source.Split> splits, long
maxSerializedBytes) {
+ List<PaimonSerializedScanTask> serializedTasks = new ArrayList<>();
+ long serializedBytes = 0;
+ for (org.apache.paimon.table.source.Split split : splits) {
+ Optional<byte[]> serializedSplit =
PaimonUtil.serializeObjectWithinLimit(
+ split, maxSerializedBytes - serializedBytes);
+ if (!serializedSplit.isPresent()) {
+ throw new PaimonTaskCacheLimitException(splits);
+ }
+ PaimonSerializedScanTask task = new
PaimonSerializedScanTask(serializedSplit.get());
+ serializedBytes += task.serializedSize();
+ serializedTasks.add(task);
+ }
+ return serializedTasks;
+ }
+
+ private static long serializedTaskBytes(List<PaimonSerializedScanTask>
tasks) {
+ return Math.max(1,
tasks.stream().mapToLong(PaimonSerializedScanTask::serializedSize).sum());
+ }
+
+ private static final class PaimonSplitTaskCacheKey
+ implements ExternalScanTaskCacheKey<PaimonSerializedScanTask> {
+ private final long catalogId;
+ private final long relationTableId;
+ private final long targetTableId;
+ private final Long snapshotId;
+ private final Long schemaId;
+ private final String scanParamType;
+ private final List<String> listParams;
+ private final Map<String, String> resolvedOptions;
+ private final Map<String, String> incrementalOptions;
+ private final Map<String, String> tableOptions;
+ private final int[] projected;
+ private final String serializedPredicates;
+
+ private PaimonSplitTaskCacheKey(
+ long catalogId, long relationTableId, long targetTableId, Long
snapshotId, Long schemaId,
+ String scanParamType, List<String> listParams,
+ Map<String, String> resolvedOptions, Map<String, String>
incrementalOptions,
+ Map<String, String> tableOptions, int[] projected,
+ String serializedPredicates) {
+ this.catalogId = catalogId;
+ this.relationTableId = relationTableId;
+ this.targetTableId = targetTableId;
+ this.snapshotId = snapshotId;
+ this.schemaId = schemaId;
+ this.scanParamType = scanParamType;
+ this.listParams = Collections.unmodifiableList(new
ArrayList<>(listParams));
+ this.resolvedOptions = Collections.unmodifiableMap(new
HashMap<>(resolvedOptions));
+ this.incrementalOptions = Collections.unmodifiableMap(new
HashMap<>(incrementalOptions));
+ this.tableOptions = Collections.unmodifiableMap(new
HashMap<>(tableOptions));
+ this.projected = Arrays.copyOf(projected, projected.length);
+ this.serializedPredicates = serializedPredicates;
+ }
+
+ @Override
+ public boolean equals(Object object) {
+ if (this == object) {
+ return true;
+ }
+ if (!(object instanceof PaimonSplitTaskCacheKey)) {
+ return false;
+ }
+ PaimonSplitTaskCacheKey that = (PaimonSplitTaskCacheKey) object;
+ return catalogId == that.catalogId
+ && relationTableId == that.relationTableId
+ && targetTableId == that.targetTableId
+ && Objects.equals(snapshotId, that.snapshotId)
+ && Objects.equals(schemaId, that.schemaId)
+ && Objects.equals(scanParamType, that.scanParamType)
+ && listParams.equals(that.listParams)
+ && resolvedOptions.equals(that.resolvedOptions)
+ && incrementalOptions.equals(that.incrementalOptions)
+ && tableOptions.equals(that.tableOptions)
+ && Arrays.equals(projected, that.projected)
+ && serializedPredicates.equals(that.serializedPredicates);
+ }
+
+ @Override
+ public int hashCode() {
+ return 31 * Objects.hash(
+ catalogId, relationTableId, targetTableId, snapshotId,
schemaId, scanParamType, listParams,
+ resolvedOptions, incrementalOptions, tableOptions,
serializedPredicates)
+ + Arrays.hashCode(projected);
+ }
+ }
+
+ private static final class PaimonSerializedScanTask {
+ private final byte[] serializedSplit;
+
+ private PaimonSerializedScanTask(byte[] serializedSplit) {
+ this.serializedSplit = serializedSplit;
+ }
+
+ private org.apache.paimon.table.source.Split deserialize() {
+ return PaimonUtil.deserializeObject(Arrays.copyOf(serializedSplit,
serializedSplit.length));
+ }
Review Comment:
[P2] Deserialize cached Paimon tasks without cloning the payload
This runs for the owner and every equivalent cache-hit alias, so
`Arrays.copyOf` allocates another plan-sized set of byte arrays on every
consumer (up to the 16 MiB retained-task budget in aggregate, and concurrent
aliases can overlap). `PaimonSerializedScanTask` owns this private final array,
and Paimon 1.4.2's `InstantiationUtil.deserializeObject` only wraps and reads
the supplied array via `ByteArrayInputStream`; it does not mutate it. Please
pass `serializedSplit` directly so reuse does not add this avoidable
allocation/GC cost.
--
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]
