github-actions[bot] commented on code in PR #69298: URL: https://github.com/apache/doris/pull/69298#discussion_r4317059450
########## fe/fe-core/src/main/java/org/apache/doris/datasource/split/FileScanRangeCompactor.java: ########## @@ -0,0 +1,136 @@ +// 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.datasource.split; + +import org.apache.doris.thrift.TFileScanRange; +import org.apache.doris.thrift.TFileScanRangeParams; +import org.apache.doris.thrift.TQueryOptions; +import org.apache.doris.thrift.TScanRangeLocations; +import org.apache.doris.thrift.TScanRangeParams; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; + +/** Reduces repeated file-scan wrappers only at the transport boundary, after scan assignment. */ +public final class FileScanRangeCompactor { + private FileScanRangeCompactor() { + } + + /** + * Compact only after fetching: maxBatchSize counts the original split descriptions, not envelopes. + * Every range comes from FileQueryScanNode.splitToScanRange, so file-specific attributes remain in + * TFileRangeDesc. Keep adjacent envelopes separate when their scan parameters or locations differ. + */ + public static List<TScanRangeLocations> compactSplitBatch(List<TScanRangeLocations> scanRanges) { + if (scanRanges.size() < 2) { + return scanRanges; + } + List<TScanRangeLocations> compacted = new ArrayList<>(); + TScanRangeLocations previous = null; + TFileScanRange previousFiles = null; + boolean copied = false; + for (TScanRangeLocations current : scanRanges) { + TFileScanRange files = current.getScanRange().getExtScanRange().getFileScanRange(); Review Comment: [P2] Update the remaining producer test envelopes for compaction SplitProducerTest.Fixture still returns TScanRangeLocations with only locations and a null scan_range (lines 593–596). Its twoScansInitializeOnOneWorkerBeforeAnyConsumerStarts test drains with getNextBatch(2). Once the consumer frees the first queue slot, the resumed worker can enqueue the next batch before the consumer polls again; the two-entry response reaches this dereference and throws NullPointerException. A slower worker hides the failure by producing one-entry responses, so the existing test becomes flaky. Update this fixture to construct file-scan envelopes, as this PR already does for SplitStreamingConcurrencyTest and PluginDrivenScanNodeHiveStreamingTest. ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java: ########## @@ -274,46 +285,61 @@ private void sweepExpiredStats(long nowNanos) { statsByQuery.entrySet().removeIf(e -> nowNanos - e.getValue().lastTouchNanos >= statsTtlNanos); } - /** - * {@code specsById} is the scan's spec map (bound to the scan's schema), the same one its residuals and delete - * index use: the reader converts partition values to these types, so they must match what the scan evaluates. - */ - private static IcebergManifestEntryKey keyOf(ManifestFile manifest, Map<Integer, PartitionSpec> specsById) { - PartitionSpec boundSpec = specsById.get(manifest.partitionSpecId()); - Preconditions.checkArgument(boundSpec != null, "No bound partition spec %s for iceberg manifest %s", - manifest.partitionSpecId(), manifest.path()); - return IcebergManifestEntryKey.of(manifest, boundSpec); - } - private static ManifestCacheValue loadManifestCacheValue(ManifestFile manifest, Table table, - Map<Integer, PartitionSpec> specsById, ManifestContent content, boolean estimateWeight) { + ManifestContent content, boolean estimateWeight) { try { if (content == ManifestContent.DELETES) { - return ManifestCacheValue.forDeleteFiles(loadDeleteFiles(manifest, table, specsById), estimateWeight); + return ManifestCacheValue.forDeleteFiles(loadDeleteFiles(manifest, table), estimateWeight); } - return ManifestCacheValue.forDataFiles(loadDataFiles(manifest, table, specsById), estimateWeight); + return ManifestCacheValue.forDataFiles(loadDataFiles(manifest, table), estimateWeight); } catch (IOException e) { throw new RuntimeException("Failed to read iceberg manifest " + manifest.path(), e); } } - private static List<DataFile> loadDataFiles(ManifestFile manifest, Table table, - Map<Integer, PartitionSpec> specsById) throws IOException { + private static List<DataFile> loadDataFiles(ManifestFile manifest, Table table) throws IOException { List<DataFile> dataFiles = new ArrayList<>(); // .copy() is mandatory: the ManifestReader iterator reuses the same object across iterations. - try (ManifestReader<DataFile> reader = ManifestFiles.read(manifest, table.io(), specsById)) { + try (ManifestReader<DataFile> reader = ManifestFiles.read(manifest, table.io(), + manifestReadSpecs(manifest, table))) { for (DataFile dataFile : reader) { dataFiles.add(dataFile.copy()); } } return dataFiles; } - private static List<DeleteFile> loadDeleteFiles(ManifestFile manifest, Table table, - Map<Integer, PartitionSpec> specsById) throws IOException { + /** Keep dropped partition sources readable for historical queries without reopening the manifest header. */ + private static Map<Integer, PartitionSpec> manifestReadSpecs(ManifestFile manifest, Table table) { + PartitionSpec spec = table.specs().get(manifest.partitionSpecId()); + boolean missingSource = spec.fields().stream() + .anyMatch(field -> spec.schema().findField(field.sourceId()) == null); + if (!missingSource) { + return Collections.singletonMap(spec.specId(), spec); + } + Map<Integer, Types.NestedField> sources = new LinkedHashMap<>(); + for (PartitionField field : spec.fields()) { + Types.NestedField source = spec.schema().findField(field.sourceId()); + if (source == null) { + source = table.schemas().entrySet().stream() + .sorted(Map.Entry.<Integer, Schema>comparingByKey().reversed()) + .map(entry -> entry.getValue().findField(field.sourceId())) + .filter(Objects::nonNull).findFirst() + .orElseThrow(() -> new IllegalArgumentException( + "Missing schema for partition source " + field.sourceId())); + } + sources.put(field.sourceId(), Types.NestedField.optional( + field.sourceId(), "source_" + field.sourceId(), source.type())); + } + PartitionSpec readable = spec.toUnbound().bind(new Schema( + new ArrayList<>(sources.values())), true); Review Comment: [P2] Avoid partition-name collisions in the recovery schema The synthetic names can invalidate a legal spec: start with p(id=1), q(id=2), partition by identity(p) and bucket(q, 16) named source_1, then remove the p partition and drop p while retaining the old manifest. Recovery introduces a schema field named source_1 with id=1, conflicting with the bucket partition named source_1 sourced from id=2. [PartitionSpec.Builder.add](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/api/src/main/java/org/apache/iceberg/PartitionSpec.java#L571-L574) still checks those name/source identities during bind(schema, true), which now throws. All historical schemas can be present here; this is separate from schema cleanup. Preserve compatible source names or otherwise prevent these collisions so cold streaming scans continue to read the retained manifest. ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestCache.java: ########## @@ -274,46 +285,61 @@ private void sweepExpiredStats(long nowNanos) { statsByQuery.entrySet().removeIf(e -> nowNanos - e.getValue().lastTouchNanos >= statsTtlNanos); } - /** - * {@code specsById} is the scan's spec map (bound to the scan's schema), the same one its residuals and delete - * index use: the reader converts partition values to these types, so they must match what the scan evaluates. - */ - private static IcebergManifestEntryKey keyOf(ManifestFile manifest, Map<Integer, PartitionSpec> specsById) { - PartitionSpec boundSpec = specsById.get(manifest.partitionSpecId()); - Preconditions.checkArgument(boundSpec != null, "No bound partition spec %s for iceberg manifest %s", - manifest.partitionSpecId(), manifest.path()); - return IcebergManifestEntryKey.of(manifest, boundSpec); - } - private static ManifestCacheValue loadManifestCacheValue(ManifestFile manifest, Table table, - Map<Integer, PartitionSpec> specsById, ManifestContent content, boolean estimateWeight) { + ManifestContent content, boolean estimateWeight) { try { if (content == ManifestContent.DELETES) { - return ManifestCacheValue.forDeleteFiles(loadDeleteFiles(manifest, table, specsById), estimateWeight); + return ManifestCacheValue.forDeleteFiles(loadDeleteFiles(manifest, table), estimateWeight); } - return ManifestCacheValue.forDataFiles(loadDataFiles(manifest, table, specsById), estimateWeight); + return ManifestCacheValue.forDataFiles(loadDataFiles(manifest, table), estimateWeight); } catch (IOException e) { throw new RuntimeException("Failed to read iceberg manifest " + manifest.path(), e); } } - private static List<DataFile> loadDataFiles(ManifestFile manifest, Table table, - Map<Integer, PartitionSpec> specsById) throws IOException { + private static List<DataFile> loadDataFiles(ManifestFile manifest, Table table) throws IOException { List<DataFile> dataFiles = new ArrayList<>(); // .copy() is mandatory: the ManifestReader iterator reuses the same object across iterations. - try (ManifestReader<DataFile> reader = ManifestFiles.read(manifest, table.io(), specsById)) { + try (ManifestReader<DataFile> reader = ManifestFiles.read(manifest, table.io(), + manifestReadSpecs(manifest, table))) { for (DataFile dataFile : reader) { dataFiles.add(dataFile.copy()); } } return dataFiles; } - private static List<DeleteFile> loadDeleteFiles(ManifestFile manifest, Table table, - Map<Integer, PartitionSpec> specsById) throws IOException { + /** Keep dropped partition sources readable for historical queries without reopening the manifest header. */ + private static Map<Integer, PartitionSpec> manifestReadSpecs(ManifestFile manifest, Table table) { + PartitionSpec spec = table.specs().get(manifest.partitionSpecId()); + boolean missingSource = spec.fields().stream() + .anyMatch(field -> spec.schema().findField(field.sourceId()) == null); + if (!missingSource) { + return Collections.singletonMap(spec.specId(), spec); + } + Map<Integer, Types.NestedField> sources = new LinkedHashMap<>(); + for (PartitionField field : spec.fields()) { + Types.NestedField source = spec.schema().findField(field.sourceId()); + if (source == null) { + source = table.schemas().entrySet().stream() + .sorted(Map.Entry.<Integer, Schema>comparingByKey().reversed()) + .map(entry -> entry.getValue().findField(field.sourceId())) + .filter(Objects::nonNull).findFirst() + .orElseThrow(() -> new IllegalArgumentException( Review Comment: [P2] Keep retained manifests readable after schema cleanup For a v2 table previously partitioned by bucket(p, 16), remove that partition field, drop p, append with the new schema, then expire the old snapshots with cleanExpiredMetadata(true). Iceberg can retain the old live manifest/spec while removing every schema containing p ([cleanup implementation](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java#L219-L249)). This lookup now throws even though the bucket partition still has a known INT result type. With the manifest cache enabled, the data-manifest read happens lazily in the streaming iterator, outside streamingFileScanTasks’ fallback, so a cold batch scan fails. Preserve readable bound specs or recover the missing type from the manifest instead of requiring its source schema to remain in table.schemas(). ########## fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergManifestPrefetcher.java: ########## @@ -0,0 +1,356 @@ +// 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.iceberg; + +import org.apache.doris.connector.spi.ConnectorContext; +import org.apache.doris.connector.spi.ConnectorTaskScheduler; + +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.io.CloseableIterator; + +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.NoSuchElementException; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.FutureTask; +import java.util.concurrent.RunnableFuture; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; +import java.util.function.Function; + +/** + * Completion-ordered, single-consumer manifest prefetch. close() may run on the query-cancellation thread. + * Only the immutable manifest-list iterator and cached-only probes run under the short registry lock; + * remote loading, future waits and cancellation all run outside it. At most window results are retained. + */ +final class IcebergManifestPrefetcher implements CloseableIterator<IcebergManifestPrefetcher.LoadedManifest> { + static final class LoadedManifest { + final ManifestFile manifest; + final ManifestCacheValue files; + + LoadedManifest(ManifestFile manifest, ManifestCacheValue files) { + this.manifest = manifest; + this.files = files; + } + } + + private static final class Pending { + final ManifestFile manifest; + final ManifestCacheValue cached; + RunnableFuture<ManifestCacheValue> future; + + Pending(ManifestFile manifest, ManifestCacheValue cached, RunnableFuture<ManifestCacheValue> future) { + this.manifest = manifest; + this.cached = cached; + this.future = future; + } + } + + private static final Pending CLOSED = new Pending(null, null, null); + + private final CloseableIterator<ManifestFile> manifests; + private final int window; + private final Function<ManifestFile, ManifestCacheValue> cached; + private final Function<ManifestFile, ManifestCacheValue> loader; + private final ConnectorContext context; + private final ConnectorTaskScheduler scheduler; + private final Semaphore readPermits; + private final BooleanSupplier allowPrefetch; + private final Deque<Pending> pending = new ArrayDeque<>(); + // Entries stay in pending until consumed, so completed results also count against the window. + // Completion callbacks never acquire the registry lock. Reserve one slot for the close signal. + private final BlockingQueue<Pending> ready; + private volatile boolean closed; + private boolean coldManifestConsumed; + + IcebergManifestPrefetcher(CloseableIterator<ManifestFile> manifests, int window, + Function<ManifestFile, ManifestCacheValue> cached, + Function<ManifestFile, ManifestCacheValue> loader, ConnectorContext context, Semaphore readPermits) { + this(manifests, window, cached, loader, context, readPermits, () -> true); + } + + IcebergManifestPrefetcher(CloseableIterator<ManifestFile> manifests, int window, + Function<ManifestFile, ManifestCacheValue> cached, + Function<ManifestFile, ManifestCacheValue> loader, ConnectorContext context, Semaphore readPermits, + BooleanSupplier allowPrefetch) { + this(manifests, window, cached, loader, context, readPermits, allowPrefetch, null); + } + + IcebergManifestPrefetcher(CloseableIterator<ManifestFile> manifests, int window, + Function<ManifestFile, ManifestCacheValue> cached, + Function<ManifestFile, ManifestCacheValue> loader, ConnectorContext context, Semaphore readPermits, + BooleanSupplier allowPrefetch, ConnectorTaskScheduler scheduler) { + this.scheduler = scheduler; + this.manifests = manifests; + this.window = window; + this.ready = new ArrayBlockingQueue<>(window + 1); + this.cached = cached; + this.loader = loader; + this.context = context; + this.readPermits = readPermits; + this.allowPrefetch = allowPrefetch; + } + + @Override + public boolean hasNext() { + synchronized (pending) { + return !closed && (!pending.isEmpty() || manifests.hasNext()); + } + } + + @Override + public LoadedManifest next() { + Pending head = null; + List<ScheduledLoad> submissions = null; + try { + synchronized (pending) { + if (closed) { + throw new CancellationException("Iceberg manifest source is closed"); + } + if (pending.isEmpty()) { + if (!manifests.hasNext()) { + throw new NoSuchElementException(); + } + ManifestFile manifest = manifests.next(); + ManifestCacheValue value = cached.apply(manifest); + if (value != null) { + // Warm cache: no futures, submissions, or speculative iterator traversal. + return new LoadedManifest(manifest, value); + } + if (window == 1 || !coldManifestConsumed || !allowPrefetch.getAsBoolean()) { + // A required serial read has no executor handoff. Expand only when useful work + // needs more metadata, without making progress depend on backend requests. + head = new Pending(manifest, null, null); + } else { + submissions = new ArrayList<>(window); + pending.addLast(submit(manifest, null, submissions)); + } + } + if (head == null) { + // If a matching file was found, pause further speculative submissions until the + // backend asks for another batch. Already-started shared reads may still complete. + while (allowPrefetch.getAsBoolean() && pending.size() < window && manifests.hasNext()) { + ManifestFile manifest = manifests.next(); + if (submissions == null) { + submissions = new ArrayList<>(window); + } + pending.addLast(submit(manifest, cached.apply(manifest), submissions)); + } + } + } + // The offline scheduler may execute inline. Even that path must never do IO under + // the pending lock: close() must be able to cancel and wake the consumer at any time. + if (submissions != null) { + submissions.forEach(ScheduledLoad::admit); + } + if (head == null) { + head = awaitReady(); + } + ManifestCacheValue value; + if (head.future == null && head.cached == null) { + value = load(head.manifest); + } else if (head.cached != null) { + value = head.cached; + } else { + // The callable publishes readiness just before FutureTask records its outcome. + // get() supplies the completion barrier and propagates failures, including Errors. + value = head.future.get(); + } + synchronized (pending) { + if (closed) { + throw new CancellationException("Iceberg manifest source is closed"); + } + if (head.future != null || head.cached != null) { + pending.remove(head); + } + coldManifestConsumed = true; + } + return new LoadedManifest(head.manifest, value); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw fail(new RuntimeException("Interrupted while loading Iceberg manifest", e)); + } catch (ExecutionException e) { + Throwable cause = e.getCause(); + throw fail(new RuntimeException("Failed to load Iceberg manifest " + head.manifest.path(), cause)); + } catch (RuntimeException e) { + throw fail(e); + } + } + + private Pending awaitReady() throws InterruptedException { + Pending help; + synchronized (pending) { + if (closed) { + throw new CancellationException("Iceberg manifest source is closed"); + } + Pending completed = ready.poll(); + if (completed != null) { + return completed; + } + help = pending.getFirst(); + } + // If every worker is occupied by a parent, run our own queued work to guarantee progress. + // If a worker already owns it, run() is a no-op: wait for ANY completion, not this future. + if (scheduler == null) { + help.future.run(); + } + Pending completed = ready.take(); + if (closed) { + throw new CancellationException("Iceberg manifest source is closed"); + } + return completed; + } + + private Pending submit(ManifestFile manifest, ManifestCacheValue value, List<ScheduledLoad> submissions) { + Pending task = new Pending(manifest, value, null); + if (value != null) { + ready.add(task); + return task; + } + if (scheduler != null) { + ScheduledLoad load = new ScheduledLoad(task); + task.future = load; + submissions.add(load); + return task; + } + Callable<ManifestCacheValue> load = () -> { + try { + return load(manifest); + } finally { + if (!closed) { + // A close racing this publication can leave at most window completions plus + // its wakeup signal. add() must never block a metadata worker on consumption. + ready.add(task); + } + } + }; + task.future = context == null ? new FutureTask<>(load) : context.submitMetadataTask(load); + return task; + } + + /** A rejected admission keeps only this window entry; it never blocks a scheduler worker. */ + private final class ScheduledLoad extends FutureTask<ManifestCacheValue> { + private final Pending pendingLoad; + private final Runnable onSpace = this::admit; + + private ScheduledLoad(Pending pendingLoad) { + super(() -> load(pendingLoad.manifest)); + this.pendingLoad = pendingLoad; + } + + private void admit() { + if (isDone()) { + return; + } + try { + scheduler.tryExecute(this, this::setException, onSpace); + } catch (RuntimeException e) { + setException(e); + } + // Cancellation can race callback registration in tryExecute. + if (isDone()) { + scheduler.removeSpaceListener(onSpace); + } + } + + @Override + protected void done() { + scheduler.removeSpaceListener(onSpace); + if (!closed) { + ready.add(pendingLoad); + } + } + } + + private ManifestCacheValue load(ManifestFile manifest) throws InterruptedException { + // Permits are acquired ONLY by a running task, never by queued futures. Consequently a waiter + // cannot depend on an unstarted permit holder when all shared-pool workers are occupied. + // Another scan of the same query can hold every permit inside remote IO. Query cleanup + // closes its group before this producer can close the source, so observe that lifetime + // while waiting as well as after admission, without interrupting a shared cache loader. + checkReadOpen(); + while (!readPermits.tryAcquire(100, TimeUnit.MILLISECONDS)) { + checkReadOpen(); + } + ClassLoader previous = Thread.currentThread().getContextClassLoader(); + try { + checkReadOpen(); + Thread.currentThread().setContextClassLoader(IcebergManifestPrefetcher.class.getClassLoader()); + if (context == null) { + return loader.apply(manifest); + } + try { + return context.executeAuthenticated(() -> loader.apply(manifest)); + } catch (RuntimeException e) { + throw e; + } catch (Exception e) { + throw new RuntimeException("Failed to load Iceberg manifest " + manifest.path(), e); + } + } finally { + Thread.currentThread().setContextClassLoader(previous); + readPermits.release(); + } + } + + private void checkReadOpen() { + if (closed || (scheduler != null && scheduler.isClosed())) { + throw new CancellationException("Iceberg manifest source is closed"); + } + } + + private RuntimeException fail(RuntimeException failure) { + try { + close(); + } catch (IOException closeFailure) { + failure.addSuppressed(closeFailure); + } + return failure; + } + + @Override + public void close() throws IOException { + List<Pending> unused; + synchronized (pending) { + if (closed) { + return; + } + closed = true; + unused = new ArrayList<>(pending); + pending.clear(); + ready.clear(); + ready.add(CLOSED); + } + for (Pending task : unused) { + if (task.future != null) { + // MetaCache single-flights this immutable load across queries. Interrupting its elected + // loader would also fail unrelated waiters. Cancel queued work, but let an ongoing shared + // read finish; closed is checked before any subsequent manifest read starts. + task.future.cancel(false); Review Comment: [P2] Retain the table owner until metadata callables finish Cancelling these futures without interruption lets their loaders keep using table.io(), but no child retains the table lease. For a REST catalog with vended credentials, the table is statement-owned: after a LIMIT query consumes a fast prefetched manifest, its producer can pause on a full split queue while another prefetch is still reading. QueryFinishCleanup then closes the statement scope and that table’s FileIO, whose [S3 implementation closes its clients](https://github.com/apache/iceberg/blob/apache-iceberg-1.12.0/aws/src/main/java/org/apache/iceberg/aws/s3/S3FileIO.java#L502-L507). A concurrent query joined to that manifest’s single-flight cache load receives the resulting failure too. Retain an owner for each running load and release it when the callable actually exits; FutureTask.done() also runs on cancel(false) and is too early. -- 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]
