github-actions[bot] commented on code in PR #69276:
URL: https://github.com/apache/doris/pull/69276#discussion_r4399167775
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/tvf/source/TVFScanNode.java:
##########
@@ -196,20 +201,17 @@ private List<Split> getLanceSplits() throws UserException
{
if (!sessionVariable.enableFileScannerV2) {
throw new UserException("Lance TVF requires
enable_file_scanner_v2=true");
}
+ List<LancePlannedDataset> datasets =
tableValuedFunction.getLancePlannedDatasets();
+ if (!datasets.isEmpty()) {
+ return LanceSplitBuilder.buildFragmentSplits(datasets, 1);
Review Comment:
[P2] Gate multi-dataset plans on BE dataset-switch support. During a smooth
upgrade, an eligible source BE can still run the previous Lance reader, whose
_ensure_dataset_open rejects a second dataset key with "Lance reader cannot mix
dataset snapshots or storage options". The existing backend check only gates
additional projected types, so a uris scan projecting a BIGINT row_id passes
it. With max_file_scanners_concurrency=1, one reader consumes ranges from both
datasets and the query fails on the second one. Check multi-dataset reader
capability before scheduling, or ensure older readers never receive more than
one dataset.
##########
be/src/format_v2/table/lance_reader.cpp:
##########
@@ -661,8 +667,16 @@ Status LanceTableReader::_open_dataset(const DatasetKey&
key) {
// EXPLAIN plans a moving selector separately, so only the profile shows
what this execution
// read. The version is the one opened, which version 0 (the latest)
resolves.
_opened_version =
static_cast<int64_t>(lance_dataset_version(dataset.get()));
- _scanner_profile->add_info_string("LanceDatasetVersion",
std::to_string(_opened_version));
- _scanner_profile->add_info_string("LanceDatasetUri", display_uri(key.uri));
+ // The profile holds one entry per dataset this reader opens, so each open
appends the URI and
+ // version it read.
+ if (!_opened_dataset_uris.empty()) {
+ _opened_dataset_uris += ", ";
+ _opened_dataset_versions += ", ";
+ }
+ _opened_dataset_uris += display_uri(key.uri);
+ _opened_dataset_versions += std::to_string(_opened_version);
+ _scanner_profile->add_info_string("LanceDatasetVersion",
_opened_dataset_versions);
+ _scanner_profile->add_info_string("LanceDatasetUri", _opened_dataset_uris);
Review Comment:
[P2] Deduplicate dataset identities before extending the profile. Fragment
splits are shuffled by FederationBackendPolicy, so a one-scanner multi-dataset
scan repeatedly reopens datasets it already visited. Each reopen appends the
same URI/version again, and both add_info_string calls copy the entire
cumulative string. A scan with many fragments therefore incurs quadratic
profile-copy work and can retain megabytes of duplicate metadata despite having
only a few distinct datasets (the loader permits 100,000 fragments). Record
each dataset URI/version once, or use a bounded summary, rather than extending
and republishing the history on every reopen.
##########
fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java:
##########
@@ -54,15 +59,26 @@ public S3TableValuedFunction(Map<String, String>
properties, List<TBrokerFileSta
throws AnalysisException {
// 1. analyze common properties
Map<String, String> props = super.parseCommonProperties(properties);
+ boolean hasUris = props.keySet().stream().anyMatch(key ->
key.equalsIgnoreCase(Uris.KEY));
Review Comment:
[P2] Use the first uris location during storage inference.
createPrimary(props) still runs before any URI from the array is available to
the existing provider/endpoint inference, which only reads the singular uri
key. For example,
s3("uris"='["https://my-bucket.s3.us-west-1.amazonaws.com/a.lance";]',
"format"="lance") fails with "No supported storage type found", although the
equivalent uri call detects S3 and derives its endpoint and region from that
URL. An explicit fs.s3.support alone still leaves endpoint/region uninferred.
Parse the list before storage construction and provide its first location to
inference, while preserving the original properties for mutual-exclusion
validation; cover the file() wrapper as well.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/metadata/LanceMultiSnapshotLoader.java:
##########
@@ -0,0 +1,288 @@
+// 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.lance.metadata;
+
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ThreadPoolManager;
+import org.apache.doris.datasource.lance.storage.LanceStorageProvider;
+import org.apache.doris.datasource.property.storage.StorageProperties;
+import org.apache.doris.datasource.property.storage.Uris;
+import org.apache.doris.qe.ConnectContext;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Opens every dataset a {@code uris} array names and pins one snapshot per
dataset.
+ *
+ * <p>Each dataset is opened on its own and keeps the version, schema and
fragments that one open
+ * read. The result is ordered the way the caller wrote the array, so the
first dataset is the
+ * reference the rest are compared against.
+ *
+ * <p>Any failure fails the whole query, and a failure names the path and the
operation.
+ */
+public final class LanceMultiSnapshotLoader {
+
+ /** The maximum number of distinct datasets, after deduplication. */
+ public static final int MAX_DATASETS = 500;
+
+ /** The maximum number of fragments, summed over every dataset. */
+ public static final long MAX_TOTAL_FRAGMENTS = 100_000L;
+
+ /** Fields in one dataset's schema. */
+ public static final int MAX_FIELDS_PER_SCHEMA = 1_000;
+
+ private static final int MAX_CONCURRENT_OPENS = 4;
+ // Pending opens allowed to wait in the executor queue.
+ private static final int MAX_QUEUED_OPENS = 4096;
+ private static final long DEADLINE_SECONDS = 60L;
+ private static final long PER_OPEN_TIMEOUT_SECONDS = 20L;
+ private static final String SCHEME_DELIMITER = "://";
+
+ private static final ThreadPoolExecutor EXECUTOR =
ThreadPoolManager.newDaemonFixedThreadPool(
+ MAX_CONCURRENT_OPENS, MAX_QUEUED_OPENS,
"lance-multi-dataset-open", true,
+ new ThreadPoolExecutor.AbortPolicy());
+
+ private LanceMultiSnapshotLoader() {
+ }
+
+ /** The datasets the plan keeps, together with the one schema it still has
to read. */
+ public static final class Loaded {
+ /** One entry per dataset, in the order written: URI, version and
fragments. */
+ public final List<LancePlannedDataset> datasets;
+ /** The schema every dataset was compared against, and the one the
plan's columns come from. */
+ public final Schema referenceSchema;
+
+ private Loaded(List<LancePlannedDataset> datasets, Schema
referenceSchema) {
+ this.datasets = datasets;
+ this.referenceSchema = referenceSchema;
+ }
+ }
+
+ /**
+ * Opens every dataset {@code normalizedUris} names.
+ *
+ * @param normalizedUris the non-empty dataset URIs to open
+ * @return the planned datasets and the schema the plan's columns come from
+ * @throws AnalysisException if a URI is not well-formed for Lance, the
datasets do not share one
+ * provider, a budget is exceeded, a dataset cannot be opened, or
two schemas differ
+ */
+ public static Loaded load(List<String> normalizedUris,
+ StorageProperties storageProperties) throws AnalysisException {
+ List<String> uris = new ArrayList<>(new
LinkedHashSet<>(normalizedUris));
+ requireWellFormed(uris);
+ requireOneProvider(uris);
+ if (uris.size() > MAX_DATASETS) {
+ throw new AnalysisException("'" + Uris.KEY + "' must not exceed "
+ MAX_DATASETS
+ + " distinct datasets, but got " + uris.size());
+ }
+
+ List<StorageProperties> properties =
Collections.singletonList(storageProperties);
+ long deadlineSeconds = deadlineSeconds();
+ long deadlineNanos = System.nanoTime() +
TimeUnit.SECONDS.toNanos(deadlineSeconds);
+
+ List<LanceTableMetadata> datasets = new ArrayList<>(uris.size());
+ for (int start = 0; start < uris.size(); start +=
MAX_CONCURRENT_OPENS) {
+ List<LanceTableMetadata> opened = openBatch(uris, start,
+ Math.min(start + MAX_CONCURRENT_OPENS, uris.size()),
properties,
+ deadlineNanos, deadlineSeconds);
+ for (LanceTableMetadata metadata : opened) {
+ requireReadable(metadata, datasets);
+ datasets.add(metadata);
+ }
+ }
+ return new Loaded(reduce(datasets), datasets.get(0).getSchema());
+ }
+
+
+ private static List<LancePlannedDataset> reduce(List<LanceTableMetadata>
datasets) {
+ List<LancePlannedDataset> planned = new ArrayList<>(datasets.size());
+ for (LanceTableMetadata metadata : datasets) {
+ planned.add(new LancePlannedDataset(
+ metadata.getDatasetUri(), metadata.getVersion(),
metadata.getFragments()));
+ }
+ return planned;
+ }
+
+ /**
+ * Checks one opened dataset against the budgets and against the reference
schema.
+ *
+ * @param opened the datasets accepted so far
+ * @throws AnalysisException if the version is invalid, the schema is
missing or too large, the
+ * fragment budget is exceeded, or the schema differs from the
reference
+ */
+ @VisibleForTesting
+ static void requireReadable(LanceTableMetadata metadata,
List<LanceTableMetadata> opened)
+ throws AnalysisException {
+ String uri = metadata.getDatasetUri();
+ if (metadata.getVersion() <= 0) {
+ throw new AnalysisException("Lance dataset '" + uri + "' returned
an invalid version "
+ + metadata.getVersion());
+ }
+ if (metadata.getSchema() == null) {
+ throw new AnalysisException("Lance dataset '" + uri + "' returned
no schema");
+ }
+ int fields = metadata.getSchema().getFields().size();
+ if (fields > MAX_FIELDS_PER_SCHEMA) {
+ throw new AnalysisException("Lance dataset '" + uri + "' has " +
fields
+ + " fields, which exceeds " + MAX_FIELDS_PER_SCHEMA);
+ }
+ long fragments = metadata.getFragments().size();
+ for (LanceTableMetadata previous : opened) {
+ fragments += previous.getFragments().size();
+ }
+ if (fragments > MAX_TOTAL_FRAGMENTS) {
+ throw new AnalysisException("The Lance datasets named by '" +
Uris.KEY + "' hold more than "
+ + MAX_TOTAL_FRAGMENTS + " fragments in total");
+ }
+ if (opened.isEmpty()) {
+ return;
+ }
+ LanceTableMetadata reference = opened.get(0);
+ Optional<String> difference = LanceSchemaEquality.firstDifference(
+ reference.getSchema(), metadata.getSchema());
+ if (difference.isPresent()) {
+ throw new AnalysisException("Lance dataset '" + uri + "' does not
have the same schema as '"
+ + reference.getDatasetUri() + "': " + difference.get()
+ + ". A Lance TVF reads only datasets whose schemas are
exactly equal.");
+ }
+ }
+
+ private static List<LanceTableMetadata> openBatch(List<String> uris, int
start, int end,
+ List<StorageProperties> properties, long deadlineNanos, long
deadlineSeconds)
+ throws AnalysisException {
+ List<Future<LanceTableMetadata>> futures = new ArrayList<>(end -
start);
+ for (int slot = start; slot < end; slot++) {
+ if (remainingNanos(deadlineNanos) <= 0) {
+ throw deadlineExceeded(deadlineSeconds, slot, uris.size());
+ }
+ final int index = slot;
+ try {
+ futures.add(EXECUTOR.submit(
+ () ->
LanceMetadataLoader.loadLatestForTvf(uris.get(index), properties)));
+ } catch (RejectedExecutionException e) {
+ throw new AnalysisException("Lance dataset open capacity is
exhausted: too many datasets "
+ + "are being opened across all queries at once; reduce
concurrency or retry");
+ }
+ }
+
+ List<LanceTableMetadata> opened = new ArrayList<>(futures.size());
+ for (int offset = 0; offset < futures.size(); offset++) {
+ String uri = uris.get(start + offset);
+ long remainingNanos = remainingNanos(deadlineNanos);
+ if (remainingNanos <= 0) {
+ throw deadlineExceeded(deadlineSeconds, start + offset,
uris.size());
+ }
+ long waitNanos =
Math.min(TimeUnit.SECONDS.toNanos(PER_OPEN_TIMEOUT_SECONDS), remainingNanos);
+ try {
+ opened.add(futures.get(offset).get(waitNanos,
TimeUnit.NANOSECONDS));
+ } catch (TimeoutException e) {
+ throw new AnalysisException("Opening Lance dataset '" + uri +
"' timed out after "
Review Comment:
[P2] Retire outstanding opens when a planning batch fails. A timeout here
abandons the futures without cancelling or removing them; deadline,
interruption, execution-failure and partial-submission-rejection exits do the
same. With all four shared workers occupied by slow storage opens, later uris
queries can time out while their work is still queued, yet those abandoned
tasks remain in the 4096-entry FIFO queue and execute after storage recovers.
Repeated failures therefore delay healthy queries and can exhaust shared
capacity after the original queries have ended. Clean up unfinished futures on
every unsuccessful exit and bound the underlying open by the planning deadline;
Future.cancel(true) alone should not be assumed to stop native I/O.
--
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]
