OIiveirra commented on code in PR #69527: URL: https://github.com/apache/doris/pull/69527#discussion_r4367889049
########## fe/fe-core/src/main/java/org/apache/doris/tablefunction/HadoopTvfAdmission.java: ########## @@ -0,0 +1,502 @@ +// 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.tablefunction; + +import org.apache.doris.common.Config; +import org.apache.doris.common.util.LocationPath; +import org.apache.doris.nereids.exceptions.AnalysisException; + +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * TVF-only admission for Hadoop consumers. Input properties have already been normalized by storage. + * This module neither loads Hadoop resources nor constructs a filesystem. Java strings and native + * C strings are projected separately before filesystem selection and mount-table interpretation. + */ +public final class HadoopTvfAdmission { + public enum Access { + NATIVE_WRITE, + JAVA_DELETE, + HADOOP_READ + } + + private enum Provider { + REMOTE, + LOCAL, + MOUNT_TABLE, + DYNAMIC, + UNKNOWN + } + + private static final Set<String> REMOTE_SCHEMES = new HashSet<>(Arrays.asList( + "hdfs", "s3a", "s3n", "oss", "ofs", "jfs", "abfs", "abfss", "wasb", "wasbs", "gs", "cosn")); + private static final Set<String> REMOTE_IMPLEMENTATIONS = new HashSet<>(Arrays.asList( + "org.apache.hadoop.hdfs.DistributedFileSystem", + "org.apache.hadoop.fs.s3a.S3AFileSystem", + "org.apache.hadoop.fs.azurebfs.AzureBlobFileSystem", + "org.apache.hadoop.fs.azurebfs.SecureAzureBlobFileSystem", + "org.apache.hadoop.fs.azure.NativeAzureFileSystem", + "org.apache.hadoop.fs.aliyun.oss.AliyunOSSFileSystem", + "com.aliyun.jindodata.oss.JindoOssFileSystem", + "org.apache.hadoop.fs.obs.OBSFileSystem", + "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem")); + + private final Access access; + private final Map<String, String> properties = new HashMap<>(); + + private HadoopTvfAdmission(Access access, Map<String, String> normalized, boolean nativeConsumer) { + this.access = access; + Set<String> nativeRoutingKeys = new HashSet<>(); + for (Map.Entry<String, String> entry : normalized.entrySet()) { + // parse_properties / generateHdfsParam consumes the canonical NameNode before JNI. + if (nativeConsumer && "fs.defaultFS".equalsIgnoreCase(entry.getKey())) { + continue; + } + String key = (nativeConsumer ? nativeStringPrefix(entry.getKey()) : entry.getKey()).trim(); + // Native writes first pass through StringCaseMap, before JNI truncates strings. + // Its comparator uses strncasecmp followed by the full UTF-8 byte length. + if (nativeConsumer && access == Access.NATIVE_WRITE && isRoutingKey(asciiCaseFold(key))) { + String identity = asciiCaseFold(nativeStringPrefix(entry.getKey())) + "\0" + + entry.getKey().getBytes(StandardCharsets.UTF_8).length; + if (!nativeRoutingKeys.add(identity)) { + reject(); + } + } + String value = nativeConsumer ? nativeStringPrefix(entry.getValue()) : entry.getValue(); + String previous = properties.put(key, value); + if (previous != null && !previous.equals(value) && isRoutingKey(key)) { + reject(); + } + } + } + + /** Reject local/unknown Hadoop TVF routes before listing, deletion, schema discovery or data IO. */ + public static void check(Access access, String path, Map<String, String> normalized) { + if (access != Access.HADOOP_READ && Config.enable_local_tvf) { + return; + } + try { + if (access == Access.NATIVE_WRITE) { + String nameNode = nativeNameNode(access, normalized); + URI output = nativeFileUri(access, path + "doris-tvf-file", nameNode); + String parent = output.getPath(); + parent = parent.substring(0, parent.lastIndexOf('/') + 1); + if (parent.length() > 1) { + parent = parent.substring(0, parent.length() - 1); + } + new HadoopTvfAdmission(access, normalized, true).checkNative(nameNode, parent); + } else { + HadoopTvfAdmission javaConsumer = new HadoopTvfAdmission(access, normalized, false); + // Project the Java consumer independently from native convert_path/JNI. + URI javaPath = hadoopPathUri(path.trim()); + javaConsumer.checkRoute(javaPath, javaPath.getPath(), new HashSet<>()); + if (access == Access.HADOOP_READ) { + // FileQueryScanNode sends this filesystem identifier in range.fs_name. BE + // falls back to hdfs_params.fs_name only when that range field is empty. + URI fileUri = LocationPath.of(path).getPath().toUri(); + String nameNode = fileUri.getScheme() + "://" + fileUri.getAuthority(); + if (nameNode.isEmpty()) { + nameNode = nativeNameNode(access, normalized); + } + HadoopTvfAdmission nativeConsumer = new HadoopTvfAdmission(access, normalized, true); + nativeConsumer.checkNativeRead(nameNode, path); + // Schema RPCs have no range.fs_name; they use THdfsParams.fs_name instead. + nativeConsumer.checkNativeRead(nativeNameNode(access, normalized), path); Review Comment: Admission now uses the parsed csvSchema state. It skips the schema NameNode only when the declared schema avoids that RPC, while preserving Java listing and native data admission. Without a declared schema, returned file statuses still receive schema-route admission because the RPC uses a returned path. Tests cover the unused file:/// default and the distinct returned-file schema fallback negative. Follow-up commit: f77cbb4a48a2327686c6494e2d382846b7fbb46c. 73 selected FE tests passed with fresh XML, no failures/errors/skips and unchanged inputs; targeted Checkstyle passed. ########## build.sh: ########## @@ -474,6 +474,49 @@ if [[ ! -f "${DORIS_THIRDPARTY}/installed/lib/${LAST_THIRDPARTY_LIB}" || fi fi +# Lance TVF storage policy needs both sides of the metadata-only SDK ABI. +# Check the selected third-party prefix only; a system Lance install cannot satisfy it. +lance_c_is_compatible() { + local install_dir="$1" + [[ ! -e "${install_dir}/lib64/.lance-installing" && + -s "${install_dir}/include/lance/lance.h" && + -s "${install_dir}/lib64/liblance_c.a" ]] || return 1 + grep -Eq '^#[[:space:]]*define[[:space:]]+LANCE_STORAGE_BASES_JSON_VERSION[[:space:]]+1([[:space:]]|$)' \ + "${install_dir}/include/lance/lance.h" || return 1 + # GNU nm and Darwin nm spell C exports differently. Read the whole stream: + # early awk exit would turn a large archive's producer SIGPIPE into failure. + "${NM:-nm}" -g "${install_dir}/lib64/liblance_c.a" 2>/dev/null | awk ' + NF >= 2 && $(NF - 1) ~ /^[TW]$/ && $NF ~ /^_?lance_dataset_storage_bases_json$/ { found = 1 } + END { exit !found } + ' +} + +lance_c_source_matches() { + local source_dir="$1" + [[ -f "${source_dir}/build-thirdparty.sh" && -f "${source_dir}/download-thirdparty.sh" && + -f "${source_dir}/vars.sh" && -f "${source_dir}/patches/lance-c-storage-bases.patch" ]] || return 1 + grep -Fq 'install_lance_c() {' "${source_dir}/build-thirdparty.sh" || return 1 + grep -Fq 'lance-c-storage-bases.patch' "${source_dir}/download-thirdparty.sh" || return 1 + cmp -s "${DORIS_HOME}/thirdparty/patches/lance-c-storage-bases.patch" \ + "${source_dir}/patches/lance-c-storage-bases.patch" || return 1 + cmp -s <(grep -E '^LANCE_C_(DOWNLOAD|NAME|SOURCE)=' "${DORIS_HOME}/thirdparty/vars.sh") \ + <(grep -E '^LANCE_C_(DOWNLOAD|NAME|SOURCE)=' "${source_dir}/vars.sh") +} + +if [[ "${BUILD_BE}" -eq 1 ]] && ! lance_c_is_compatible "${DORIS_THIRDPARTY}/installed"; then + if ! lance_c_source_matches "${DORIS_THIRDPARTY}"; then + echo "Lance SDK header/archive pair lacks LANCE_STORAGE_BASES_JSON_VERSION=1 or its exported API, or is incomplete." >&2 + echo "Refresh the compilation image's include/lance and lib64/liblance_c.a together, or set DORIS_THIRDPARTY to matching complete third-party sources." >&2 + exit 1 + fi + echo "Rebuilding only lance_c for the TVF storage-base metadata ABI ..." + bash "${DORIS_THIRDPARTY}/build-thirdparty.sh" -j "${PARALLEL}" lance_c Review Comment: The ABI repair now rebuilds lance_c and paimon_rust together from matching source, passing one selected cargo command to both and preserving the selected Rust toolchain environment. Conflicting library-specific cargo overrides fail before rebuilding. Existing compatible installed pairs remain reusable and unrelated dependencies are preserved. All 12 build-gate tests pass; this follow-up did not rebuild actual Rust artifacts or claim a new full BE link. Follow-up commit: f77cbb4a48a2327686c6494e2d382846b7fbb46c. 73 selected FE tests passed with fresh XML, no failures/errors/skips and unchanged inputs; targeted Checkstyle passed. -- 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]
