LanceSnapshotResolver.java

// 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.lance.Version;

import java.util.Collection;
import java.util.Comparator;
import java.util.List;
import java.util.regex.Pattern;

/** Resolves Doris time-travel selectors to immutable Lance version IDs. */
public final class LanceSnapshotResolver {
    private LanceSnapshotResolver() {
    }

    public static long parseVersion(String value) {
        final long version;
        try {
            version = Long.parseLong(value);
        } catch (NumberFormatException e) {
            // Deliberately not chained: the catalog reports the root cause message, and the
            // NumberFormatException text would replace this one.
            throw new IllegalArgumentException(isVersionNumber(value)
                    ? "Lance FOR VERSION AS OF version " + value + " is out of range"
                    : "Lance FOR VERSION AS OF requires a numeric version, but was '" + value + "'");
        }
        if (version <= 0) {
            throw new IllegalArgumentException(
                    "Lance FOR VERSION AS OF requires a positive version, but was " + version);
        }
        return version;
    }

    private static final Pattern VERSION_NUMBER = Pattern.compile("[+-]?[0-9]+");

    /**
     * Whether a {@code FOR VERSION AS OF} value is a version number rather than a tag name. A
     * signed number counts as one, so that {@code '-1'} is reported as an invalid version.
     */
    public static boolean isVersionNumber(String value) {
        return VERSION_NUMBER.matcher(value).matches();
    }

    /** No version of a chain was committed at or before the requested {@code FOR TIME AS OF} time. */
    public static final class NoVersionAtOrBeforeException extends IllegalArgumentException {
        private NoVersionAtOrBeforeException(String requestedText) {
            super("Lance dataset has no version at or before '" + requestedText + "'");
        }
    }

    /** The commit time a manifest records, at millisecond precision. */
    public static long commitMillis(Version version) {
        return version.getDataTime().toInstant().toEpochMilli();
    }

    static long versionAtOrBefore(List<Version> versions, long timestampMillis) {
        return versionAtOrBefore(versions, timestampMillis, String.valueOf(timestampMillis));
    }

    /**
     * Selects the latest version whose manifest commit time does not exceed the requested
     * timestamp, compared at millisecond precision, as Lance resolves {@code asof}.
     *
     * @param requestedText the user's {@code FOR TIME AS OF} text, echoed in the error message
     * @throws NoVersionAtOrBeforeException if every version was committed after the timestamp
     */
    public static long versionAtOrBefore(Collection<Version> versions, long timestampMillis, String requestedText) {
        return versions.stream()
                .filter(version -> commitMillis(version) <= timestampMillis)
                .max(Comparator.comparingLong(LanceSnapshotResolver::commitMillis).thenComparingLong(Version::getId))
                .orElseThrow(() -> new NoVersionAtOrBeforeException(requestedText))
                .getId();
    }
}