zy-kkk commented on code in PR #68453:
URL: https://github.com/apache/doris/pull/68453#discussion_r4129038120
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceCatalogClient.java:
##########
@@ -235,68 +250,382 @@ public LanceTableMetadata loadBasicTableMetadata(String
dbName, String tableName
}
public Schema loadTableSchema(String dbName, String tableName) {
- return readTableSnapshot(dbName, tableName, Optional.empty(),
+ return readTableSnapshot(dbName, tableName, LanceRefSelector.latest(),
(dataset, access, metrics) -> metrics.measure(Stage.SCHEMA,
dataset::getSchema));
}
public LanceTableMetadata loadTableMetadata(String dbName, String
tableName,
Optional<TableSnapshot> tableSnapshot) {
- return loadQueryMetadata(dbName, tableName, tableSnapshot,
LanceMetadataLoader.MetadataScope.WITH_INDEXES);
+ return loadTableMetadata(dbName, tableName,
LanceRefSelector.snapshot(tableSnapshot));
+ }
+
+ public LanceTableMetadata loadTableMetadata(String dbName, String
tableName, LanceRefSelector selector) {
+ return loadQueryMetadata(dbName, tableName, selector,
LanceMetadataLoader.MetadataScope.WITH_INDEXES);
}
private LanceTableMetadata loadQueryMetadata(String dbName, String
tableName,
Optional<TableSnapshot> tableSnapshot,
LanceMetadataLoader.MetadataScope mode) {
- return readTableSnapshot(dbName, tableName, tableSnapshot,
+ return loadQueryMetadata(dbName, tableName,
LanceRefSelector.snapshot(tableSnapshot), mode);
+ }
+
+ private LanceTableMetadata loadQueryMetadata(String dbName, String
tableName,
+ LanceRefSelector selector, LanceMetadataLoader.MetadataScope mode)
{
+ return readTableSnapshot(dbName, tableName, selector,
(dataset, access, metrics) ->
LanceMetadataLoader.read(dataset, access, mode, metrics));
}
- /** Pins one resource generation, resolved table access, and the Dataset
version for the whole read. */
- private <T> T readTableSnapshot(String dbName, String tableName,
Optional<TableSnapshot> tableSnapshot,
+ /**
+ * Pins one resource generation, resolved table access, and the Dataset
version for the whole read.
+ *
+ * <p>The latest version of the main chain is opened once and every other
selector is a
+ * checkout from that handle, so the SDK resolves the ref with the same
commit handler
+ * (the namespace's, for a managed table). A tag is resolved first to the
chain and version it
+ * points at, so a tag created on a branch selects that branch. The two
shortcuts that skip the
+ * latest open are an explicit version on the main chain, and {@code FOR
TIME AS OF} on a
+ * managed table whose namespace reports commit times.
+ */
+ private <T> T readTableSnapshot(String dbName, String tableName,
LanceRefSelector selector,
SnapshotReader<T> reader) {
- LanceTableAccess tableAccess = null;
+ ReadState state = new ReadState(selector, dbName + "." + tableName);
LanceMetadataMetrics metrics =
LanceMetadataMetrics.startMetadataRead();
try {
T result;
try (BufferAllocator allocator =
namespaceAllocator.newChildAllocator(
"lance-metadata-read", 0, namespaceAllocator.getLimit())) {
- tableAccess = metrics.measure(Stage.TABLE_ACCESS,
+ state.access = metrics.measure(Stage.TABLE_ACCESS,
() -> namespaceClient.resolveTableAccess(dbName,
tableName));
- OptionalLong version = OptionalLong.empty();
- if (tableSnapshot.isPresent()) {
- TableSnapshot snapshot = tableSnapshot.get();
- if (snapshot.getType() ==
TableSnapshot.VersionType.VERSION) {
- version =
OptionalLong.of(LanceSnapshotResolver.parseVersion(snapshot.getValue()));
- } else {
- long timestamp =
TimeUtils.timeStringToLong(snapshot.getValue(), TimeUtils.getTimeZone());
- if (timestamp < 0) {
- throw new IllegalArgumentException(
- "Cannot parse Lance FOR TIME AS OF value
'" + snapshot.getValue() + "'");
- }
- try (Dataset latest = openDataset(allocator,
tableAccess, OptionalLong.empty(), metrics)) {
- version =
OptionalLong.of(metrics.measure(Stage.VERSION_RESOLVE,
- () ->
LanceSnapshotResolver.getVersionAtOrBefore(latest, timestamp)));
- }
+ OptionalLong direct = directMainVersion(state, metrics);
+ if (direct.isPresent() || isLatestMain(selector)) {
+ state.version = direct;
+ try (Dataset dataset = openDataset(allocator,
state.access, direct, metrics)) {
+ result = reader.read(dataset, state.access, metrics);
+ }
+ } else {
+ try (Dataset main = openDataset(allocator, state.access,
OptionalLong.empty(), metrics)) {
+ result = readFromLatest(main, state, reader, metrics);
}
- }
- try (Dataset dataset = openDataset(allocator, tableAccess,
version, metrics)) {
- result = reader.read(dataset, tableAccess, metrics);
}
}
metrics.succeeded();
return result;
+ } catch (LanceUserFacingException e) {
+ throw new RuntimeException(e.getMessage(), e);
} catch (Exception e) {
- throw LanceErrorMessages.failure("Failed to load Lance table
metadata for " + dbName + "." + tableName, e,
- tableAccess == null ? null : tableAccess.getDatasetUri(),
- tableAccess == null ? namespaceStorageOptions :
tableAccess.getStorageOptions(), catalogSecrets);
+ LanceTableAccess access = state.access;
+ String uri = access == null ? null : access.getDatasetUri();
+ Map<String, String> options = access == null ?
namespaceStorageOptions : access.getStorageOptions();
+ String what = state.displayName();
+ if (state.branch.isPresent() && !state.branchCheckedOut &&
isBranchNotFound(e, state.branch.get())) {
+ throw new RuntimeException("Lance branch '" +
state.branch.get() + "' of " + state.tableName
+ + state.selector.getTag().map(tag -> " (tag '" + tag +
"')").orElse("")
+ + " was not found" + (isNamespaceMiss(e, "table branch
not found") ? " in the namespace" : ""),
+ sanitizedCause(e, uri, options));
+ }
+ if (state.version.isPresent() && isVersionNotFound(e)) {
+ throw new RuntimeException("Lance version " +
state.version.getAsLong() + " of " + what
+ + state.selector.getTag().map(tag -> " (tag '" + tag +
"')").orElse("")
+ + " was not found" + (isNamespaceMiss(e, "table
version not found") ? " in the namespace" : ""),
+ sanitizedCause(e, uri, options));
+ }
+ String hint = access != null && access.isManagedVersioning() &&
isAccessDenied(e)
+ ? " (reading a namespace-managed Lance table may need
write access to finalize a staged manifest)"
+ : "";
+ throw LanceErrorMessages.failure("Failed to load Lance table
metadata for " + what + hint, e, uri, options,
+ catalogSecrets);
} finally {
metrics.close();
}
}
+ /** What a read has resolved so far; the catch block reports errors
against it. */
+ private static final class ReadState {
+ private final LanceRefSelector selector;
+ private final String tableName;
+ private LanceTableAccess access;
+ private Optional<String> branch;
+ /** Set once the branch's latest version was checked out, i.e. the
branch exists. */
+ private boolean branchCheckedOut;
+ private OptionalLong version = OptionalLong.empty();
+ /** The namespace's version list per chain ("" is main), fetched at
most once per read. */
+ private final Map<String, List<TableVersion>> namespaceVersions = new
HashMap<>();
+
+ private ReadState(LanceRefSelector selector, String tableName) {
+ this.selector = selector;
+ this.tableName = tableName;
+ this.branch = selector.getBranch();
+ }
+
+ private String displayName() {
+ return tableName + branch.map(name -> "@" + name).orElse("");
+ }
+ }
+
+ private static boolean isLatestMain(LanceRefSelector selector) {
+ return !selector.getTag().isPresent() &&
!selector.getBranch().isPresent()
+ && !selector.getSnapshot().isPresent();
+ }
+
+ /**
+ * The main-chain version a selector names without looking at the latest
manifest: an explicit
+ * version, or {@code FOR TIME AS OF} on a managed table whose namespace
reports commit times.
+ */
+ private OptionalLong directMainVersion(ReadState state,
LanceMetadataMetrics metrics) {
+ LanceRefSelector selector = state.selector;
+ if (selector.getTag().isPresent() || selector.getBranch().isPresent()
|| !selector.getSnapshot().isPresent()) {
+ return OptionalLong.empty();
+ }
+ TableSnapshot snapshot = selector.getSnapshot().get();
+ if (snapshot.getType() == TableSnapshot.VersionType.VERSION) {
+ return
OptionalLong.of(LanceSnapshotResolver.parseVersion(snapshot.getValue()));
+ }
+ if (!state.access.isManagedVersioning()) {
+ return OptionalLong.empty();
+ }
+ long timestamp = parseTimeTravelTimestamp(snapshot.getValue());
+ return LanceSnapshotResolver.namespaceVersionAtOrBefore(
+ namespaceVersions(state, state.access, metrics), timestamp,
snapshot.getValue());
+ }
+
+ /** Resolves the selector against the open latest main chain and reads the
selected snapshot. */
+ private <T> T readFromLatest(Dataset main, ReadState state,
SnapshotReader<T> reader, LanceMetadataMetrics metrics)
+ throws Exception {
+ LanceRefSelector selector = state.selector;
+ if (selector.getTag().isPresent()) {
+ String tag = selector.getTag().get();
+ Tag target = metrics.measure(Stage.VERSION_RESOLVE, () ->
main.tags().list().stream()
+ .filter(candidate ->
tag.equals(candidate.getName())).findFirst()
+ .orElseThrow(() -> new LanceUserFacingException(
+ "Lance tag '" + tag + "' of " + state.tableName +
" was not found")));
+ state.branch = target.getBranch().filter(name ->
!MAIN_BRANCH.equals(name));
+ state.version = OptionalLong.of(target.getVersion());
+ if (!state.branch.isPresent()) {
+ try (Dataset dataset = checkout(main,
Ref.ofMain(target.getVersion()), metrics)) {
+ return reader.read(dataset, state.access, metrics);
+ }
+ }
+ }
+ if (state.branch.isPresent()) {
+ String branch = state.branch.get();
+ // Check out the branch's latest version first even when a version
is already known, so
+ // a missing branch and a missing version inside an existing
branch are told apart.
+ try (Dataset latest = checkout(main, Ref.ofBranch(branch),
metrics)) {
+ state.branchCheckedOut = true;
+ LanceTableAccess branchAccess = accessOf(latest, state);
+ if (!state.version.isPresent() &&
selector.getSnapshot().isPresent()) {
+ state.version = resolveSnapshotVersion(latest,
branchAccess, selector.getSnapshot().get(), state,
+ metrics);
+ }
+ if (!state.version.isPresent()) {
+ return reader.read(latest, branchAccess, metrics);
+ }
+ try (Dataset dataset = checkout(latest, Ref.ofBranch(branch,
state.version.getAsLong()), metrics)) {
+ return reader.read(dataset, branchAccess, metrics);
+ }
+ }
+ }
+ // FOR TIME AS OF on the main chain, resolved from storage commit
times.
+ state.version = resolveSnapshotVersion(main, state.access,
selector.getSnapshot().get(), state, metrics);
+ try (Dataset dataset = checkout(main,
Ref.ofMain(state.version.getAsLong()), metrics)) {
+ return reader.read(dataset, state.access, metrics);
+ }
+ }
+
+ /**
+ * The access for a dataset checked out from the table: the main chain
keeps the table access,
+ * and a branch takes the directory the SDK checked out, which is what the
BE opens by URI.
+ */
+ private static LanceTableAccess accessOf(Dataset dataset, ReadState state)
{
+ return state.branch.isPresent() ?
state.access.onBranch(state.branch.get(), dataset.uri()) : state.access;
+ }
+
+ /** A selector error whose message is user-facing as is, such as a tag
that does not exist. */
+ private static final class LanceUserFacingException extends
RuntimeException {
+ private LanceUserFacingException(String message) {
+ super(message);
+ }
+ }
+
+ private RuntimeException sanitizedCause(Throwable error, String uri,
Map<String, String> options) {
+ return new RuntimeException(LanceErrorMessages.sanitize(error, uri,
options, catalogSecrets));
+ }
+
+ /**
+ * Checks out a ref of an already open dataset. The SDK resolves the ref
itself, from the
+ * dataset directory or, for a namespace-managed dataset, with its own
namespace client.
+ */
+ private static Dataset checkout(Dataset dataset, Ref ref,
LanceMetadataMetrics metrics) {
+ return metrics.measure(Stage.VERSION_RESOLVE, () ->
dataset.checkout(ref));
+ }
+
+ /**
+ * Resolves a {@code FOR VERSION AS OF} / {@code FOR TIME AS OF} snapshot
against the chain
+ * {@code latest} is checked out on: the main chain, or a branch when
{@code access} is a
+ * branch access.
+ */
+ private OptionalLong resolveSnapshotVersion(Dataset latest,
LanceTableAccess access, TableSnapshot snapshot,
+ ReadState state, LanceMetadataMetrics metrics) {
+ if (snapshot.getType() == TableSnapshot.VersionType.VERSION) {
+ return
OptionalLong.of(LanceSnapshotResolver.parseVersion(snapshot.getValue()));
+ }
+ long timestamp = parseTimeTravelTimestamp(snapshot.getValue());
+ try {
+ return OptionalLong.of(resolveVersionAtOrBefore(latest, access,
timestamp, snapshot.getValue(), state,
+ metrics));
+ } catch (IllegalArgumentException e) {
+ if (!access.getBranch().isPresent()) {
+ throw e;
+ }
+ // A branch's chain starts at the version it was created from and
carries its own
+ // commit times, so an earlier timestamp has nothing to select on
the branch.
+ throw new LanceUserFacingException("Lance branch '" +
access.getBranch().get() + "' of "
+ + state.tableName + " has no version at or before '" +
snapshot.getValue()
+ + "'; a branch only holds the versions from its creation
on");
+ }
+ }
+
+ /**
+ * Whether a failed branch checkout means the branch does not exist. The
SDK reports
+ * "branch <name> does not exist", a namespace "Table branch not found",
or a missing manifest
+ * under the branch directory when nothing was ever committed there.
+ */
+ private static boolean isBranchNotFound(Throwable throwable, String
branch) {
+ if (ExceptionUtils.indexOfType(throwable,
TableBranchNotFoundException.class) >= 0) {
+ return true;
+ }
+ String rootMessage = ExceptionUtils.getRootCauseMessage(throwable);
+ if (rootMessage == null) {
+ return false;
+ }
+ String lower = rootMessage.toLowerCase(Locale.ROOT);
+ String name = branch.toLowerCase(Locale.ROOT);
+ return lower.contains("table branch not found")
+ || lower.contains("branch " + name + " does not exist")
+ || (lower.contains("not found") && lower.contains("tree/" +
name + "/"));
+ }
+
+ /**
+ * Whether a not-found came from the namespace rather than storage. The
SDK surfaces a
+ * namespace error by its display text ("Table version not found: ..."),
and the Java client
+ * by its exception type.
+ */
+ private static boolean isNamespaceMiss(Throwable throwable, String
namespaceText) {
+ if (ExceptionUtils.indexOfType(throwable,
TableVersionNotFoundException.class) >= 0
+ || ExceptionUtils.indexOfType(throwable,
TableBranchNotFoundException.class) >= 0) {
+ return true;
+ }
+ String rootMessage = ExceptionUtils.getRootCauseMessage(throwable);
+ return rootMessage != null &&
rootMessage.toLowerCase(Locale.ROOT).contains(namespaceText);
+ }
+
+ /** An HTTP 403 as the object stores report it, or an explicit
access-denied error. */
+ private static final Pattern ACCESS_DENIED = Pattern.compile(
+ "accessdenied|access denied|permission
denied|forbidden|(status|http|code)\\W{0,3}403\\b");
+
+ private static boolean isAccessDenied(Throwable throwable) {
+ String rootMessage = ExceptionUtils.getRootCauseMessage(throwable);
+ return rootMessage != null &&
ACCESS_DENIED.matcher(rootMessage.toLowerCase(Locale.ROOT)).find();
+ }
+
+ /**
+ * Every version the namespace records for the chain {@code access}
addresses, listed once per
+ * read. The whole list is needed: the storage fallback filters by it, and
neither the order a
+ * namespace returns nor monotonic commit times can be relied on to stop
early.
+ */
+ private List<TableVersion> namespaceVersions(ReadState state,
LanceTableAccess access,
+ LanceMetadataMetrics metrics) {
+ return
state.namespaceVersions.computeIfAbsent(access.getBranch().orElse(""), chain ->
{
+ List<TableVersion> versions =
metrics.measure(Stage.VERSION_RESOLVE,
+ () -> namespaceClient.listManagedVersions(access));
+ if (versions.isEmpty()) {
+ throw new LanceUserFacingException("Lance namespace lists no
versions for "
+ + state.tableName + (chain.isEmpty() ? "" : "@" +
chain));
+ }
+ return versions;
+ });
+ }
+
+ /**
+ * Resolves {@code FOR TIME AS OF} to a version on the chain {@code
latest} is checked out on.
+ * A namespace-managed table is resolved from the commit times the
namespace records, so that
+ * only versions the namespace knows are selected. If the namespace lists
its versions without
+ * commit times, the times come from the manifests present in storage,
restricted to the
+ * versions the namespace lists; a storage-versioned table is resolved
from storage alone.
+ */
+ private long resolveVersionAtOrBefore(Dataset latest, LanceTableAccess
access, long timestamp,
+ String requestedText, ReadState state, LanceMetadataMetrics
metrics) {
+ Set<Long> recordedVersions = null;
+ if (access.isManagedVersioning()) {
+ List<TableVersion> recorded = namespaceVersions(state, access,
metrics);
+ OptionalLong fromNamespace =
LanceSnapshotResolver.namespaceVersionAtOrBefore(
+ recorded, timestamp, requestedText);
+ if (fromNamespace.isPresent()) {
+ LOG.debug("Resolved Lance FOR TIME AS OF '{}' to version {}
from the namespace",
+ requestedText, fromNamespace.getAsLong());
+ return fromNamespace.getAsLong();
+ }
+ recordedVersions =
recorded.stream().map(TableVersion::getVersion).collect(Collectors.toSet());
+ }
+ Set<Long> allowedVersions = recordedVersions;
+ long version = metrics.measure(Stage.VERSION_RESOLVE, () -> {
+ List<Version> versions = latest.listVersions();
Review Comment:
Update after the redesign (see the PR comment): a recorded version the
storage listing lacks is no longer checked out to finalize it. It cuts the `FOR
TIME AS OF` history like a removed version; when the namespace records it at a
staged manifest, the error says the version cannot be read.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceCatalogClient.java:
##########
@@ -235,68 +252,508 @@ public LanceTableMetadata loadBasicTableMetadata(String
dbName, String tableName
}
public Schema loadTableSchema(String dbName, String tableName) {
- return readTableSnapshot(dbName, tableName, Optional.empty(),
+ return readTableSnapshot(dbName, tableName, LanceRefSelector.latest(),
(dataset, access, metrics) -> metrics.measure(Stage.SCHEMA,
dataset::getSchema));
}
public LanceTableMetadata loadTableMetadata(String dbName, String
tableName,
Optional<TableSnapshot> tableSnapshot) {
- return loadQueryMetadata(dbName, tableName, tableSnapshot,
LanceMetadataLoader.MetadataScope.WITH_INDEXES);
+ return loadTableMetadata(dbName, tableName,
LanceRefSelector.snapshot(tableSnapshot));
+ }
+
+ public LanceTableMetadata loadTableMetadata(String dbName, String
tableName, LanceRefSelector selector) {
+ return loadQueryMetadata(dbName, tableName, selector,
LanceMetadataLoader.MetadataScope.WITH_INDEXES);
}
private LanceTableMetadata loadQueryMetadata(String dbName, String
tableName,
Optional<TableSnapshot> tableSnapshot,
LanceMetadataLoader.MetadataScope mode) {
- return readTableSnapshot(dbName, tableName, tableSnapshot,
+ return loadQueryMetadata(dbName, tableName,
LanceRefSelector.snapshot(tableSnapshot), mode);
+ }
+
+ private LanceTableMetadata loadQueryMetadata(String dbName, String
tableName,
+ LanceRefSelector selector, LanceMetadataLoader.MetadataScope mode)
{
+ return readTableSnapshot(dbName, tableName, selector,
(dataset, access, metrics) ->
LanceMetadataLoader.read(dataset, access, mode, metrics));
}
- /** Pins one resource generation, resolved table access, and the Dataset
version for the whole read. */
- private <T> T readTableSnapshot(String dbName, String tableName,
Optional<TableSnapshot> tableSnapshot,
+ /**
+ * Pins one resource generation, resolved table access, and the Dataset
version for the whole read.
+ *
+ * <p>The latest version of the main chain is opened once and every other
selector is a
+ * checkout from that handle, so the SDK resolves the ref with the same
commit handler
+ * (the namespace's, for a managed table). A tag is resolved first to the
chain and version it
+ * points at, so a tag created on a branch selects that branch. The two
shortcuts that skip the
+ * latest open are an explicit version on the main chain, and the latest
version of a managed
+ * table. For a managed table, "latest" is always the newest version the
namespace records,
+ * never the newest manifest in storage.
+ */
+ private <T> T readTableSnapshot(String dbName, String tableName,
LanceRefSelector selector,
SnapshotReader<T> reader) {
- LanceTableAccess tableAccess = null;
+ ReadState state = new ReadState(selector, dbName + "." + tableName);
LanceMetadataMetrics metrics =
LanceMetadataMetrics.startMetadataRead();
try {
T result;
try (BufferAllocator allocator =
namespaceAllocator.newChildAllocator(
"lance-metadata-read", 0, namespaceAllocator.getLimit())) {
- tableAccess = metrics.measure(Stage.TABLE_ACCESS,
+ state.access = metrics.measure(Stage.TABLE_ACCESS,
() -> namespaceClient.resolveTableAccess(dbName,
tableName));
- OptionalLong version = OptionalLong.empty();
- if (tableSnapshot.isPresent()) {
- TableSnapshot snapshot = tableSnapshot.get();
- if (snapshot.getType() ==
TableSnapshot.VersionType.VERSION) {
- version =
OptionalLong.of(LanceSnapshotResolver.parseVersion(snapshot.getValue()));
- } else {
- long timestamp =
TimeUtils.timeStringToLong(snapshot.getValue(), TimeUtils.getTimeZone());
- if (timestamp < 0) {
- throw new IllegalArgumentException(
- "Cannot parse Lance FOR TIME AS OF value
'" + snapshot.getValue() + "'");
- }
- try (Dataset latest = openDataset(allocator,
tableAccess, OptionalLong.empty(), metrics)) {
- version =
OptionalLong.of(metrics.measure(Stage.VERSION_RESOLVE,
- () ->
LanceSnapshotResolver.getVersionAtOrBefore(latest, timestamp)));
- }
+ OptionalLong direct = directMainVersion(state, metrics);
+ if (direct.isPresent() || isLatestMain(selector)) {
+ state.version = direct;
+ try (Dataset dataset = openDataset(allocator, state,
direct, metrics)) {
+ result = reader.read(dataset, state.access, metrics);
+ }
+ } else {
+ OptionalLong mainVersion =
state.access.isManagedVersioning()
+ ? OptionalLong.of(recordedLatestVersion(state,
Optional.empty(), metrics))
+ : OptionalLong.empty();
+ try (Dataset main = openDataset(allocator, state,
mainVersion, metrics)) {
+ result = readFromLatest(main, state, reader, metrics);
}
- }
- try (Dataset dataset = openDataset(allocator, tableAccess,
version, metrics)) {
- result = reader.read(dataset, tableAccess, metrics);
}
}
metrics.succeeded();
return result;
- } catch (Exception e) {
- throw LanceErrorMessages.failure("Failed to load Lance table
metadata for " + dbName + "." + tableName, e,
- tableAccess == null ? null : tableAccess.getDatasetUri(),
- tableAccess == null ? namespaceStorageOptions :
tableAccess.getStorageOptions(), catalogSecrets);
+ } catch (LanceUserFacingException e) {
+ throw new RuntimeException(e.getMessage(), e);
+ } catch (Exception sdkError) {
+ Exception e = unwrapCallbackFailure(state, sdkError);
+ LanceTableAccess access = state.access;
+ String uri = access == null ? null : access.getDatasetUri();
+ Map<String, String> options = access == null ?
namespaceStorageOptions : access.getStorageOptions();
+ String what = state.displayName();
+ if (state.branch.isPresent() && !state.branchExists &&
isBranchNotFound(e, state.branch.get())) {
+ throw new RuntimeException("Lance branch '" +
state.branch.get() + "' of " + state.tableName
+ + state.selector.getTag().map(tag -> " (tag '" + tag +
"')").orElse("")
+ + " was not found" + (isNamespaceMiss(e, "table branch
not found") ? " in the namespace" : ""),
+ sanitizedCause(e, uri, options));
+ }
+ if (state.version.isPresent() && isVersionNotFound(e)) {
+ throw new RuntimeException("Lance version " +
state.version.getAsLong() + " of " + what
+ + state.selector.getTag().map(tag -> " (tag '" + tag +
"')").orElse("")
+ + " was not found" + (isNamespaceMiss(e, "table
version not found") ? " in the namespace" : ""),
+ sanitizedCause(e, uri, options));
+ }
+ String hint = access != null && access.isManagedVersioning() &&
isAccessDenied(e)
+ ? " (reading a namespace-managed Lance table may need
write access to finalize a staged manifest)"
+ : "";
+ throw LanceErrorMessages.failure("Failed to load Lance table
metadata for " + what + hint, e, uri, options,
+ catalogSecrets);
} finally {
metrics.close();
}
}
- private Dataset openDataset(BufferAllocator allocator, LanceTableAccess
access, OptionalLong version,
+ /** What a read has resolved so far; the catch block reports errors
against it. */
+ private static final class ReadState {
+ private final LanceRefSelector selector;
+ private final String tableName;
+ private LanceTableAccess access;
+ /** The namespace the SDK opened a managed table through, which keeps
its callbacks' failures. */
+ private LanceSdkNamespace sdkNamespace;
+ private Optional<String> branch;
+ /**
+ * Set once the branch is known to exist: the namespace recorded
versions for it, or its
+ * latest version was checked out. Later failures are not reported as
a missing branch.
+ */
+ private boolean branchExists;
+ private OptionalLong version = OptionalLong.empty();
+ /** The namespace's version list per chain ("" is main), fetched at
most once per read. */
+ private final Map<String, List<TableVersion>> namespaceVersions = new
HashMap<>();
+
+ private ReadState(LanceRefSelector selector, String tableName) {
+ this.selector = selector;
+ this.tableName = tableName;
+ this.branch = selector.getBranch();
+ }
+
+ private String displayName() {
+ return tableName + branch.map(name -> "@" + name).orElse("");
+ }
+ }
+
+ private static boolean isLatestMain(LanceRefSelector selector) {
+ return !selector.getTag().isPresent() &&
!selector.getBranch().isPresent()
+ && !selector.getSnapshot().isPresent();
+ }
+
+ /**
+ * The main-chain version a selector names without looking at the latest
manifest: an explicit
+ * version, or the latest version of a managed table, which the namespace
records.
+ */
+ private OptionalLong directMainVersion(ReadState state,
LanceMetadataMetrics metrics) {
+ LanceRefSelector selector = state.selector;
+ if (selector.getTag().isPresent() || selector.getBranch().isPresent())
{
+ return OptionalLong.empty();
+ }
+ if (!selector.getSnapshot().isPresent()) {
+ return state.access.isManagedVersioning()
+ ? OptionalLong.of(recordedLatestVersion(state,
Optional.empty(), metrics))
+ : OptionalLong.empty();
+ }
+ TableSnapshot snapshot = selector.getSnapshot().get();
+ return snapshot.getType() == TableSnapshot.VersionType.VERSION
+ ?
OptionalLong.of(LanceSnapshotResolver.parseVersion(snapshot.getValue()))
+ : OptionalLong.empty();
+ }
+
+ /** Resolves the selector against the open latest main chain and reads the
selected snapshot. */
+ private <T> T readFromLatest(Dataset main, ReadState state,
SnapshotReader<T> reader, LanceMetadataMetrics metrics)
+ throws Exception {
+ LanceRefSelector selector = state.selector;
+ if (selector.getTag().isPresent()) {
+ // Only this tag's file is read, however many tags the table has.
The SDK checks the tag
+ // out on the branch of the version it points at; for a managed
table that is an
+ // explicit version the namespace resolves, never a storage
fallback.
+ String tag = selector.getTag().get();
+ state.version =
OptionalLong.of(metrics.measure(Stage.VERSION_RESOLVE, () -> tagVersion(main,
tag, state)));
+ try (Dataset target = checkout(main, Ref.ofTag(tag), metrics)) {
+ state.branch = branchOf(target.uri(),
state.access.getDatasetUri());
+ return reader.read(target, accessOf(target, state), metrics);
+ }
+ }
+ if (state.branch.isPresent()) {
+ String branch = state.branch.get();
+ // Check out the branch's latest version first even when a version
is already known, so
+ // a missing branch and a missing version inside an existing
branch are told apart.
+ Ref branchHead = Ref.ofBranch(branch);
+ if (state.access.isManagedVersioning()) {
+ branchHead = Ref.ofBranch(branch, recordedLatestVersion(state,
Optional.of(branch), metrics));
+ state.branchExists = true;
+ }
+ try (Dataset latest = checkout(main, branchHead, metrics)) {
+ state.branchExists = true;
+ LanceTableAccess branchAccess = accessOf(latest, state);
+ if (!state.version.isPresent() &&
selector.getSnapshot().isPresent()) {
+ state.version = resolveSnapshotVersion(latest,
branchAccess, selector.getSnapshot().get(), state,
+ metrics);
+ }
+ if (!state.version.isPresent()) {
+ return reader.read(latest, branchAccess, metrics);
+ }
+ try (Dataset dataset = checkout(latest, Ref.ofBranch(branch,
state.version.getAsLong()), metrics)) {
+ return reader.read(dataset, branchAccess, metrics);
+ }
+ }
+ }
+ // FOR TIME AS OF on the main chain, resolved from manifest commit
times.
+ state.version = resolveSnapshotVersion(main, state.access,
selector.getSnapshot().get(), state, metrics);
+ try (Dataset dataset = checkout(main,
Ref.ofMain(state.version.getAsLong()), metrics)) {
+ return reader.read(dataset, state.access, metrics);
+ }
+ }
+
+ private static long tagVersion(Dataset main, String tag, ReadState state) {
+ try {
+ return main.tags().getVersion(tag);
+ } catch (RuntimeException e) {
+ String rootMessage = ExceptionUtils.getRootCauseMessage(e);
+ if (rootMessage != null && rootMessage.contains("tag " + tag + "
does not exist")) {
+ throw new LanceUserFacingException("Lance tag '" + tag + "' of
" + state.tableName + " was not found");
+ }
+ throw e;
+ }
+ }
+
+ /**
+ * The branch a dataset checked out from the table root is on, from its
root directory: the
+ * table root for main, {@code <root>/tree/<branch>} otherwise. Lance
inserts the branch path
+ * before a URI's query string, so the query is compared apart. A URI that
is neither is an
+ * error rather than main, which would hand the BE the wrong chain.
+ */
+ static Optional<String> branchOf(String checkedOutUri, String tableUri) {
+ String root =
StringUtils.removeEnd(StringUtils.substringBefore(tableUri, "?"), "/");
+ String uri =
StringUtils.removeEnd(StringUtils.substringBefore(checkedOutUri, "?"), "/");
+ if (uri.equals(root)) {
+ return Optional.empty();
+ }
+ String branchRoot = root + "/tree/";
+ if (!uri.startsWith(branchRoot) || uri.length() ==
branchRoot.length()) {
+ // The URIs may carry credentials in their query, so they stay out
of the message.
+ throw new IllegalStateException("Cannot tell which branch a Lance
tag was checked out on");
+ }
+ return Optional.of(uri.substring(branchRoot.length()));
+ }
+
+ /**
+ * The access for a dataset checked out from the table: the main chain
keeps the table access,
+ * and a branch takes the directory the SDK checked out, which is what the
BE opens by URI.
+ */
+ private static LanceTableAccess accessOf(Dataset dataset, ReadState state)
{
+ return state.branch.isPresent() ?
state.access.onBranch(state.branch.get(), dataset.uri()) : state.access;
+ }
+
+ /** A selector error whose message is user-facing as is, such as a tag
that does not exist. */
+ private static final class LanceUserFacingException extends
RuntimeException {
+ private LanceUserFacingException(String message) {
+ super(message);
+ }
+ }
+
+ /**
+ * The newest version the namespace records for a managed chain. Doris
asks for it itself:
+ * opening "latest" through the SDK falls back to the newest manifest in
storage when the
+ * namespace records none, which would expose a version the namespace
never published.
+ */
+ private long recordedLatestVersion(ReadState state, Optional<String>
branch, LanceMetadataMetrics metrics) {
+ // A read that already listed the chain's versions reuses that list.
+ List<TableVersion> listed =
state.namespaceVersions.get(branch.orElse(""));
+ OptionalLong latest = listed != null
+ ?
listed.stream().map(TableVersion::getVersion).filter(Objects::nonNull)
+ .mapToLong(Long::longValue).max()
+ : metrics.measure(Stage.VERSION_RESOLVE,
+ () ->
namespaceClient.latestManagedVersion(state.access, branch));
+ if (!latest.isPresent()) {
+ throw new LanceUserFacingException("Lance namespace lists no
versions for " + state.tableName
+ + branch.map(name -> "@" + name).orElse(""));
+ }
+ return latest.getAsLong();
+ }
+
+ /**
+ * The failure behind {@code sdkError}. For a managed table the SDK
resolves versions through
+ * {@link LanceSdkNamespace} by JNI callback, and reports a namespace
error there without its
+ * type or message; the namespace kept it.
+ */
+ private static Exception unwrapCallbackFailure(ReadState state, Exception
sdkError) {
+ return state.sdkNamespace == null ? sdkError :
state.sdkNamespace.unwrapCallbackFailure(sdkError);
+ }
+
+ private RuntimeException sanitizedCause(Throwable error, String uri,
Map<String, String> options) {
+ return new RuntimeException(LanceErrorMessages.sanitize(error, uri,
options, catalogSecrets));
+ }
+
+ /**
+ * Checks out a ref of an already open dataset. The SDK resolves the ref
itself, from the
+ * dataset directory or, for a namespace-managed dataset, with its own
namespace client.
+ */
+ private static Dataset checkout(Dataset dataset, Ref ref,
LanceMetadataMetrics metrics) {
+ return metrics.measure(Stage.VERSION_RESOLVE, () ->
dataset.checkout(ref));
+ }
+
+ /**
+ * Resolves a {@code FOR VERSION AS OF} / {@code FOR TIME AS OF} snapshot
against the chain
+ * {@code latest} is checked out on: the main chain, or a branch when
{@code access} is a
+ * branch access.
+ */
+ private OptionalLong resolveSnapshotVersion(Dataset latest,
LanceTableAccess access, TableSnapshot snapshot,
+ ReadState state, LanceMetadataMetrics metrics) {
+ if (snapshot.getType() == TableSnapshot.VersionType.VERSION) {
+ return
OptionalLong.of(LanceSnapshotResolver.parseVersion(snapshot.getValue()));
+ }
+ long timestamp = parseTimeTravelTimestamp(snapshot.getValue());
+ try {
+ return OptionalLong.of(resolveVersionAtOrBefore(latest, access,
timestamp, snapshot.getValue(), state,
+ metrics));
+ } catch (LanceSnapshotResolver.NoVersionAtOrBeforeException e) {
+ if (!access.getBranch().isPresent()) {
+ throw e;
+ }
+ // A branch's chain starts at the version it was created from and
carries its own
+ // commit times, so an earlier timestamp has nothing to select on
the branch.
+ throw new LanceUserFacingException("Lance branch '" +
access.getBranch().get() + "' of "
+ + state.tableName + " has no version at or before '" +
snapshot.getValue()
+ + "'; a branch only holds the versions from its creation
on");
+ }
+ }
+
+ /**
+ * Whether a failed branch checkout means the branch does not exist. The
SDK reports
+ * "branch <name> does not exist", a namespace "Table branch not found",
or a missing manifest
+ * under the branch directory when nothing was ever committed there.
+ */
+ private static boolean isBranchNotFound(Throwable throwable, String
branch) {
+ if (ExceptionUtils.indexOfType(throwable,
TableBranchNotFoundException.class) >= 0) {
+ return true;
+ }
+ String rootMessage = ExceptionUtils.getRootCauseMessage(throwable);
+ if (rootMessage == null) {
+ return false;
+ }
+ String lower = rootMessage.toLowerCase(Locale.ROOT);
+ String name = branch.toLowerCase(Locale.ROOT);
+ return lower.contains("table branch not found")
+ || lower.contains("branch " + name + " does not exist")
+ || (lower.contains("not found") && lower.contains("tree/" +
name + "/"));
+ }
+
+ /**
+ * Whether a not-found came from the namespace rather than storage. The
SDK surfaces a
+ * namespace error by its display text ("Table version not found: ..."),
and the Java client
+ * by its exception type.
+ */
+ private static boolean isNamespaceMiss(Throwable throwable, String
namespaceText) {
+ if (ExceptionUtils.indexOfType(throwable,
TableVersionNotFoundException.class) >= 0
+ || ExceptionUtils.indexOfType(throwable,
TableBranchNotFoundException.class) >= 0) {
+ return true;
+ }
+ String rootMessage = ExceptionUtils.getRootCauseMessage(throwable);
+ return rootMessage != null &&
rootMessage.toLowerCase(Locale.ROOT).contains(namespaceText);
+ }
+
+ /** An HTTP 403 as the object stores report it, or an explicit
access-denied error. */
+ private static final Pattern ACCESS_DENIED = Pattern.compile(
+ "accessdenied|access denied|permission
denied|forbidden|(status|http|code)\\W{0,3}403\\b");
+
+ private static boolean isAccessDenied(Throwable throwable) {
+ String rootMessage = ExceptionUtils.getRootCauseMessage(throwable);
+ return rootMessage != null &&
ACCESS_DENIED.matcher(rootMessage.toLowerCase(Locale.ROOT)).find();
+ }
+
+ /**
+ * Every version the namespace records for the chain {@code access}
addresses, listed once per
+ * read. The whole list is needed: the storage fallback filters by it, and
neither the order a
+ * namespace returns nor monotonic commit times can be relied on to stop
early.
+ */
+ private List<TableVersion> namespaceVersions(ReadState state,
LanceTableAccess access,
LanceMetadataMetrics metrics) {
+ return
state.namespaceVersions.computeIfAbsent(access.getBranch().orElse(""), chain ->
{
+ List<TableVersion> versions =
metrics.measure(Stage.VERSION_RESOLVE,
+ () -> namespaceClient.listManagedVersions(access));
+ if (versions.isEmpty()) {
+ throw new LanceUserFacingException("Lance namespace lists no
versions for "
+ + state.tableName + (chain.isEmpty() ? "" : "@" +
chain));
+ }
+ return versions;
+ });
+ }
+
+ /**
+ * Resolves {@code FOR TIME AS OF} to a version on the chain {@code
latest} is checked out on,
+ * from the commit times the manifests record, over the history {@link
LanceSnapshotResolver}
+ * describes. A managed table only selects among the versions its
namespace records; one
+ * missing from the storage listing because its manifest is still staged
is checked out, which
+ * finalizes it and yields its commit time.
+ */
+ private long resolveVersionAtOrBefore(Dataset latest, LanceTableAccess
access, long timestamp,
+ String requestedText, ReadState state, LanceMetadataMetrics
metrics) {
+ NavigableSet<Long> recorded = access.isManagedVersioning()
+ ? namespaceVersions(state, access,
metrics).stream().map(TableVersion::getVersion)
+
.filter(Objects::nonNull).collect(Collectors.toCollection(TreeSet::new))
+ : null;
+ long version = metrics.measure(Stage.VERSION_RESOLVE, () -> {
+ try {
+ return
LanceSnapshotResolver.versionAtOrBefore(latest.listVersions(), recorded,
+ id -> recordedVersion(latest, access, id, state),
timestamp, requestedText);
+ } catch (LanceSnapshotResolver.HistoryRemovedException e) {
+ throw historyRemoved(e.getVersion(), requestedText, state);
+ }
+ });
+ LOG.debug("Resolved Lance FOR TIME AS OF '{}' to version {} from
manifest commit times", requestedText,
+ version);
+ return version;
+ }
+
+ /**
+ * A namespace-recorded version checked out through the namespace, or null
if it is gone. A
+ * still-staged manifest is finalized by the checkout.
+ */
+ private static Version recordedVersion(Dataset latest, LanceTableAccess
access, long version,
+ ReadState state) {
+ Ref ref = access.getBranch().map(name -> Ref.ofBranch(name,
version)).orElseGet(() -> Ref.ofMain(version));
+ try (Dataset recorded = latest.checkout(ref)) {
+ return recorded.getVersion();
+ } catch (Exception e) {
+ // Also the IOException the JNI raises for a missing manifest.
+ Exception failure = unwrapCallbackFailure(state, e);
+ if (isVersionNotFound(failure)) {
+ return null;
+ }
+ // The namespace's own exception if it kept one; otherwise the
SDK's, which may be the
+ // checked IOException the JNI throws undeclared.
+ if (failure instanceof RuntimeException) {
+ throw (RuntimeException) failure;
+ }
+ throw e;
+ }
+ }
+
+ private static LanceUserFacingException historyRemoved(long version,
String requestedText, ReadState state) {
+ return new LanceUserFacingException("Lance cannot resolve FOR TIME AS
OF '" + requestedText + "' on "
+ + state.displayName() + ": version " + version + ", which may
hold the state at that time,"
+ + " no longer exists");
+ }
+
+ /**
+ * Parses a {@code FOR TIME AS OF} value in the session time zone. Second
and millisecond
+ * precision are accepted; commit times are compared at millisecond
precision, the precision a
+ * namespace reports them in, so a timestamp in the millisecond a commit
lands in selects it.
+ */
+ private static long parseTimeTravelTimestamp(String value) {
+ long timestamp = TimeUtils.timeStringToLong(value,
TimeUtils.getTimeZone());
+ if (timestamp < 0) {
+ timestamp = TimeUtils.msTimeStringToLong(value,
TimeUtils.getTimeZone());
+ }
+ if (timestamp < 0) {
+ throw new IllegalArgumentException("Cannot parse Lance FOR TIME AS
OF value '" + value
+ + "', expected 'yyyy-MM-dd HH:mm:ss' or 'yyyy-MM-dd
HH:mm:ss.SSS'");
+ }
+ return timestamp;
+ }
+
+ /**
+ * Whether a failed open of an explicitly requested version means that
version does not exist.
+ * A namespace reports it through {@link TableVersionNotFoundException}.
The storage reader
+ * reports it as a missing manifest under {@code _versions/} or as Lance's
own version-not-found
+ * error; a missing dataset or an unreachable store fails differently and
keeps its message.
+ */
+ private static boolean isVersionNotFound(Throwable throwable) {
+ if (ExceptionUtils.indexOfType(throwable,
TableVersionNotFoundException.class) >= 0) {
+ return true;
+ }
+ String rootMessage = ExceptionUtils.getRootCauseMessage(throwable);
+ if (rootMessage == null) {
+ return false;
+ }
+ String lower = rootMessage.toLowerCase(Locale.ROOT);
+ return lower.contains("version not found")
+ || (lower.contains("not found") &&
lower.contains("_versions/"));
+ }
+
+ /**
+ * Opens the main chain of the table {@code state} resolved. For a managed
table the SDK
+ * describes the table again and opens the location and storage options
that describe returns,
+ * so {@code state.access} is replaced by the access for what it opened:
the BE reads with the
+ * access this read ends up with, and must open what the FE planned. If
the SDK did not open
+ * with exactly that access's options, the dataset is opened once more
with them. A namespace
+ * that returns a relative location cannot be read in this mode.
+ */
+ private Dataset openDataset(BufferAllocator allocator, ReadState state,
OptionalLong version,
+ LanceMetadataMetrics metrics) {
+ if (state.access.isManagedVersioning()) {
+ LanceTableAccess access = state.access;
+ for (int attempt = 0; ; attempt++) {
+ ReadOptions readOptions =
LanceReadOptions.forSharedSession(access.getStorageOptions(), version,
+ session);
+ LanceTableAccess requested = access;
+ LanceSdkNamespace sdkNamespace =
namespaceClient.sdkNamespace(requested);
+ state.sdkNamespace = sdkNamespace;
+ Dataset dataset = metrics.measure(Stage.DATASET_OPEN, () ->
namespaceClient.openManagedDataset(
Review Comment:
Update after the redesign: the check moved to `LanceManifestPaths`, since
the SDK wrapper is gone. A version recorded at a staged manifest is read only
when its canonical manifest exists; Doris copies nothing. Tests:
`LanceManifestPathsTest` (runs on CI) and
`LanceManagedVersioningTest.testFinalizedManifestOutsideItsCanonicalPathFailsTheRead`.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceCatalogClient.java:
##########
@@ -235,68 +252,508 @@ public LanceTableMetadata loadBasicTableMetadata(String
dbName, String tableName
}
public Schema loadTableSchema(String dbName, String tableName) {
- return readTableSnapshot(dbName, tableName, Optional.empty(),
+ return readTableSnapshot(dbName, tableName, LanceRefSelector.latest(),
(dataset, access, metrics) -> metrics.measure(Stage.SCHEMA,
dataset::getSchema));
}
public LanceTableMetadata loadTableMetadata(String dbName, String
tableName,
Optional<TableSnapshot> tableSnapshot) {
- return loadQueryMetadata(dbName, tableName, tableSnapshot,
LanceMetadataLoader.MetadataScope.WITH_INDEXES);
+ return loadTableMetadata(dbName, tableName,
LanceRefSelector.snapshot(tableSnapshot));
+ }
+
+ public LanceTableMetadata loadTableMetadata(String dbName, String
tableName, LanceRefSelector selector) {
+ return loadQueryMetadata(dbName, tableName, selector,
LanceMetadataLoader.MetadataScope.WITH_INDEXES);
}
private LanceTableMetadata loadQueryMetadata(String dbName, String
tableName,
Optional<TableSnapshot> tableSnapshot,
LanceMetadataLoader.MetadataScope mode) {
- return readTableSnapshot(dbName, tableName, tableSnapshot,
+ return loadQueryMetadata(dbName, tableName,
LanceRefSelector.snapshot(tableSnapshot), mode);
+ }
+
+ private LanceTableMetadata loadQueryMetadata(String dbName, String
tableName,
+ LanceRefSelector selector, LanceMetadataLoader.MetadataScope mode)
{
+ return readTableSnapshot(dbName, tableName, selector,
(dataset, access, metrics) ->
LanceMetadataLoader.read(dataset, access, mode, metrics));
}
- /** Pins one resource generation, resolved table access, and the Dataset
version for the whole read. */
- private <T> T readTableSnapshot(String dbName, String tableName,
Optional<TableSnapshot> tableSnapshot,
+ /**
+ * Pins one resource generation, resolved table access, and the Dataset
version for the whole read.
+ *
+ * <p>The latest version of the main chain is opened once and every other
selector is a
+ * checkout from that handle, so the SDK resolves the ref with the same
commit handler
+ * (the namespace's, for a managed table). A tag is resolved first to the
chain and version it
+ * points at, so a tag created on a branch selects that branch. The two
shortcuts that skip the
+ * latest open are an explicit version on the main chain, and the latest
version of a managed
+ * table. For a managed table, "latest" is always the newest version the
namespace records,
+ * never the newest manifest in storage.
+ */
+ private <T> T readTableSnapshot(String dbName, String tableName,
LanceRefSelector selector,
SnapshotReader<T> reader) {
- LanceTableAccess tableAccess = null;
+ ReadState state = new ReadState(selector, dbName + "." + tableName);
LanceMetadataMetrics metrics =
LanceMetadataMetrics.startMetadataRead();
try {
T result;
try (BufferAllocator allocator =
namespaceAllocator.newChildAllocator(
"lance-metadata-read", 0, namespaceAllocator.getLimit())) {
- tableAccess = metrics.measure(Stage.TABLE_ACCESS,
+ state.access = metrics.measure(Stage.TABLE_ACCESS,
() -> namespaceClient.resolveTableAccess(dbName,
tableName));
- OptionalLong version = OptionalLong.empty();
- if (tableSnapshot.isPresent()) {
- TableSnapshot snapshot = tableSnapshot.get();
- if (snapshot.getType() ==
TableSnapshot.VersionType.VERSION) {
- version =
OptionalLong.of(LanceSnapshotResolver.parseVersion(snapshot.getValue()));
- } else {
- long timestamp =
TimeUtils.timeStringToLong(snapshot.getValue(), TimeUtils.getTimeZone());
- if (timestamp < 0) {
- throw new IllegalArgumentException(
- "Cannot parse Lance FOR TIME AS OF value
'" + snapshot.getValue() + "'");
- }
- try (Dataset latest = openDataset(allocator,
tableAccess, OptionalLong.empty(), metrics)) {
- version =
OptionalLong.of(metrics.measure(Stage.VERSION_RESOLVE,
- () ->
LanceSnapshotResolver.getVersionAtOrBefore(latest, timestamp)));
- }
+ OptionalLong direct = directMainVersion(state, metrics);
Review Comment:
Update: the behavior changed with the redesign. A managed read describes the
table once and checks every version it gets from the namespace against that
location's canonical paths, so a move between the describe and the version
lookup now fails the read with a hint to retry, instead of re-reading at the
new head.
`LanceManagedVersioningTest.testTableMovedAfterItWasDescribedFailsTheRead`
covers the failure and the retried read; the tests listed above went away with
the SDK open. The recorded path is relative to its bucket, as the namespace
records it, so a move to another bucket under the same path is not seen; the
read then uses the version the namespace named at the old location.
##########
fe/fe-core/src/test/java/org/apache/doris/datasource/lance/LanceManagedS3StoreTest.java:
##########
@@ -0,0 +1,412 @@
+// 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;
+
+import org.apache.doris.common.util.JsonUtil;
+import org.apache.doris.datasource.lance.metadata.LanceTableMetadata;
+
+import com.google.common.io.ByteStreams;
+import com.sun.jna.Library;
+import com.sun.jna.Native;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.lance.Dataset;
+import org.lance.Fragment;
+import org.lance.FragmentMetadata;
+import org.lance.FragmentOperation;
+import org.lance.WriteParams;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.math.BigInteger;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Stream;
+
+/**
+ * Reads a namespace-managed table on S3 through two S3 stubs that serve
different data under one
+ * key, standing in for two endpoints of a table. The namespace vends the
endpoint in its own
+ * spelling ({@code endpoint}), as a REST catalog may.
+ *
+ * <p>The FE must plan from the endpoint the BE is handed. That breaks if the
SDK reuses the object
+ * store another read still holds for the same table, or if the FE
environment's
+ * {@code AWS_ENDPOINT} takes the place of the vended endpoint.
+ */
+@Disabled("Re-enable after fixing Arrow C Data JNI compatibility: CI libstdc++
lacks CXXABI_1.3.9")
Review Comment:
Update: `LanceSdkNamespaceTest`, `LanceSdkOpenedAccessTest` and
`LanceManagedS3StoreTest` were removed with the SDK open. On CI now:
`LanceManifestPathsTest`, `LanceSnapshotTest` and `LanceTableAccessCacheTest`
without JNI, and `test_lance_rest_time_travel` with eight managed fixture
tables (new: `_pending`, `_staged`, `_misrecorded`). Locally, every Lance test
class with the `@Disabled`s lifted passes.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/storage/LanceStorageOptions.java:
##########
@@ -68,6 +68,26 @@ public static Map<String, String>
fromDorisAndVendedStorageOptions(String datase
return buildStorageOptions(datasetUri, storageProperties,
vendedOptions);
}
+ /**
+ * The options to hand the Lance SDK when it opens a namespace-managed
table itself. The SDK
+ * describes the table again and adds what the namespace vends then, in
the namespace's own
+ * spelling. So the vended options are handed over in that spelling too,
in place of their
+ * normalized twins in {@code merged}: the SDK's fresh values then replace
them key for key,
+ * a namespace that vends new credentials on every describe cannot leave
the SDK with a key
+ * from one describe and a secret from the other, and if that describe
vends nothing the
+ * SDK still has these.
+ */
+ public static Map<String, String> forManagedSdkOpen(String datasetUri,
Map<String, String> merged,
+ Map<String, String> vendedOptions) {
+ Map<String, String> result = new HashMap<>(merged);
+ if (vendedOptions != null && !vendedOptions.isEmpty()) {
+
result.keySet().removeAll(LanceStorageProvider.forDataset(datasetUri)
+ .normalizeVendedStorageOptions(vendedOptions).keySet());
+ result.putAll(vendedOptions);
Review Comment:
Superseded by the redesign (see the PR comment): the FE no longer opens
managed tables through the SDK's namespace client, so the SDK-open path that
handed Lance raw vended aliases is gone. Both readers now open the dataset by
URI with the options Doris builds from the read's single `DescribeTable`, where
a vended `endpoint` is already normalized to the canonical `aws_endpoint`, so
`with_env_s3` does not fill it from `AWS_ENDPOINT`. For a URI open Lance keys
the native store by all of its options, so the store-sharing concern of the
earlier thread does not apply either.
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/lance/LanceCatalogClient.java:
##########
@@ -235,68 +250,382 @@ public LanceTableMetadata loadBasicTableMetadata(String
dbName, String tableName
}
public Schema loadTableSchema(String dbName, String tableName) {
- return readTableSnapshot(dbName, tableName, Optional.empty(),
+ return readTableSnapshot(dbName, tableName, LanceRefSelector.latest(),
(dataset, access, metrics) -> metrics.measure(Stage.SCHEMA,
dataset::getSchema));
}
public LanceTableMetadata loadTableMetadata(String dbName, String
tableName,
Optional<TableSnapshot> tableSnapshot) {
- return loadQueryMetadata(dbName, tableName, tableSnapshot,
LanceMetadataLoader.MetadataScope.WITH_INDEXES);
+ return loadTableMetadata(dbName, tableName,
LanceRefSelector.snapshot(tableSnapshot));
+ }
+
+ public LanceTableMetadata loadTableMetadata(String dbName, String
tableName, LanceRefSelector selector) {
+ return loadQueryMetadata(dbName, tableName, selector,
LanceMetadataLoader.MetadataScope.WITH_INDEXES);
}
private LanceTableMetadata loadQueryMetadata(String dbName, String
tableName,
Optional<TableSnapshot> tableSnapshot,
LanceMetadataLoader.MetadataScope mode) {
- return readTableSnapshot(dbName, tableName, tableSnapshot,
+ return loadQueryMetadata(dbName, tableName,
LanceRefSelector.snapshot(tableSnapshot), mode);
+ }
+
+ private LanceTableMetadata loadQueryMetadata(String dbName, String
tableName,
+ LanceRefSelector selector, LanceMetadataLoader.MetadataScope mode)
{
+ return readTableSnapshot(dbName, tableName, selector,
(dataset, access, metrics) ->
LanceMetadataLoader.read(dataset, access, mode, metrics));
}
- /** Pins one resource generation, resolved table access, and the Dataset
version for the whole read. */
- private <T> T readTableSnapshot(String dbName, String tableName,
Optional<TableSnapshot> tableSnapshot,
+ /**
+ * Pins one resource generation, resolved table access, and the Dataset
version for the whole read.
+ *
+ * <p>The latest version of the main chain is opened once and every other
selector is a
+ * checkout from that handle, so the SDK resolves the ref with the same
commit handler
+ * (the namespace's, for a managed table). A tag is resolved first to the
chain and version it
+ * points at, so a tag created on a branch selects that branch. The two
shortcuts that skip the
+ * latest open are an explicit version on the main chain, and {@code FOR
TIME AS OF} on a
+ * managed table whose namespace reports commit times.
+ */
+ private <T> T readTableSnapshot(String dbName, String tableName,
LanceRefSelector selector,
SnapshotReader<T> reader) {
- LanceTableAccess tableAccess = null;
+ ReadState state = new ReadState(selector, dbName + "." + tableName);
LanceMetadataMetrics metrics =
LanceMetadataMetrics.startMetadataRead();
try {
T result;
try (BufferAllocator allocator =
namespaceAllocator.newChildAllocator(
"lance-metadata-read", 0, namespaceAllocator.getLimit())) {
- tableAccess = metrics.measure(Stage.TABLE_ACCESS,
+ state.access = metrics.measure(Stage.TABLE_ACCESS,
() -> namespaceClient.resolveTableAccess(dbName,
tableName));
- OptionalLong version = OptionalLong.empty();
- if (tableSnapshot.isPresent()) {
- TableSnapshot snapshot = tableSnapshot.get();
- if (snapshot.getType() ==
TableSnapshot.VersionType.VERSION) {
- version =
OptionalLong.of(LanceSnapshotResolver.parseVersion(snapshot.getValue()));
- } else {
- long timestamp =
TimeUtils.timeStringToLong(snapshot.getValue(), TimeUtils.getTimeZone());
- if (timestamp < 0) {
- throw new IllegalArgumentException(
- "Cannot parse Lance FOR TIME AS OF value
'" + snapshot.getValue() + "'");
- }
- try (Dataset latest = openDataset(allocator,
tableAccess, OptionalLong.empty(), metrics)) {
- version =
OptionalLong.of(metrics.measure(Stage.VERSION_RESOLVE,
- () ->
LanceSnapshotResolver.getVersionAtOrBefore(latest, timestamp)));
- }
+ OptionalLong direct = directMainVersion(state, metrics);
+ if (direct.isPresent() || isLatestMain(selector)) {
+ state.version = direct;
+ try (Dataset dataset = openDataset(allocator,
state.access, direct, metrics)) {
+ result = reader.read(dataset, state.access, metrics);
Review Comment:
Superseded by the redesign (see the PR comment): the FE no longer opens
managed tables through the SDK's namespace client, so `LanceSdkNamespace` and
this mechanism are gone. Both readers now open the dataset by URI with the
options of the read's single `DescribeTable`; for a URI open Lance keys the
native store by all of its options, credentials included.
--
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]
