github-actions[bot] commented on code in PR #66729:
URL: https://github.com/apache/doris/pull/66729#discussion_r4003487943
##########
fe/be-java-extensions/jdbc-scanner/src/main/java/org/apache/doris/jdbc/JdbcJniScanner.java:
##########
@@ -330,17 +334,25 @@ public void close() throws IOException {
}
@Override
- public Map<String, String> getStatistics() {
+ protected Map<String, String> collectStatistics() {
Map<String, String> stats = new HashMap<>();
stats.put("counter:ReadRows", String.valueOf(readRows));
stats.put("timer:ReadTime", String.valueOf(readTime));
return stats;
}
- private void initializeClassLoaderAndDataSource() throws Exception {
- java.net.URL[] urls = {new java.net.URL(jdbcDriverUrl)};
- ClassLoader parent = getClass().getClassLoader();
- this.classLoader = java.net.URLClassLoader.newInstance(urls, parent);
+ private void initializeClassLoaderAndDataSource() {
+ // Before the driver classes are loaded, because that is when they
read these. Both
+ // properties this sets are there to keep a driver's own cleanup
thread from pinning the
+ // classloader that loaded it - which matters more now than it did,
not less: every plugin
+ // has a classloader of its own, and a driver thread holding one keeps
a whole plugin alive.
+ typeHandler.setSystemProperties();
+ // The checksum the catalog was defined with, checked once per driver
jar - when its
+ // classloader is created, not on every scan. A driver jar replaced in
place at the same
+ // URL is otherwise served from the cache until BE restarts, and a jar
that is not the one
+ // the catalog names is a silent wrong answer rather than an error.
+ this.classLoader = JdbcDriverUtils.driverClassLoader(jdbcDriverUrl,
getClass().getClassLoader(),
Review Comment:
[P1] Retire the pool when the driver checksum changes
`driverClassLoader` correctly verifies the new checksum and drops the cached
loader when a jar is replaced at the same URL, but the following datasource
lookup is keyed by `JdbcDataSource.createCacheKey`, which does not include
`jdbcDriverChecksum`. An existing Hikari pool therefore survives the ALTER and
still owns the old Driver instance/pooled connections. The connection tester
uses a separate temporary pool, so validation may pass against the new jar
while scans and writes continue through the old one. Include the checksum (or a
driver-loader generation) in the pool identity and retire the old pool safely.
##########
build.sh:
##########
@@ -1376,57 +1452,132 @@ EOF
cp -r -p "${DORIS_HOME}/be/output/lib/task_executor_simulator"
"${DORIS_OUTPUT}/be/lib/"/
fi
- extensions_modules=("java-udf")
- extensions_modules+=("jdbc-scanner")
- extensions_modules+=("hadoop-hudi-scanner")
- extensions_modules+=("paimon-scanner")
- extensions_modules+=("trino-connector-scanner")
- extensions_modules+=("max-compute-connector")
- # lakesoul-scanner has been deprecated
- # extensions_modules+=("lakesoul-scanner")
- extensions_modules+=("preload-extensions")
- extensions_modules+=("iceberg-metadata-scanner")
- extensions_modules+=("${HADOOP_DEPS_NAME}")
- extensions_modules+=("java-writer")
+ # Everything from here to the end of this block deploys what the Java
extension build
+ # produced, so it only runs when there was one.
DISABLE_BE_JAVA_EXTENSIONS=ON (and the
+ # Darwin fallback that sets the same flag when JAVA_HOME has no usable
libjvm) leaves every
+ # target/ below empty, and the plugin loop is a hard failure when a jar is
missing - which
+ # is how a BE-only build, .github/workflows/be-ut-mac.yml included, died
here.
+ if [[ "${BUILD_BE_JAVA_EXTENSIONS}" -eq 1 ]]; then
+
+ # Every be-java-extensions module that BE addresses by name is a
plugin now. The one exception
+ # is the hadoop drop below, which is not a plugin and never was, so
there is no list of
+ # "extensions modules" left to iterate - only that one flag.
+ deploy_hadoop_deps=1
+ if [[ -n "${BE_EXTENSION_IGNORE}" ]]; then
+ IFS=',' read -r -a ignore_modules <<<"${BE_EXTENSION_IGNORE}"
+ for ignore_module in "${ignore_modules[@]}"; do
+ if [[ "${ignore_module// /}" == "${HADOOP_DEPS_NAME}" ]]; then
+ deploy_hadoop_deps=0
+ break
+ fi
+ done
+ fi
- if [[ -n "${BE_EXTENSION_IGNORE}" ]]; then
- IFS=',' read -r -a ignore_modules <<<"${BE_EXTENSION_IGNORE}"
- new_modules=()
- for module in "${extensions_modules[@]}"; do
- module=${module// /}
- if [[ -n "${module}" ]]; then
+ # The shared layer: the SPI a plugin compiles against and the loader
that reads the plugin
+ # directory. These are the only Doris classes that live on both sides
of the boundary, which
+ # is why they are the only ones deployed where the system classpath
can see them.
+ BE_JAVA_SPI_DIR="${DORIS_OUTPUT}/be/lib/jni/spi"
+ rm -rf "${DORIS_OUTPUT}/be/lib/jni"
+ mkdir -p "${BE_JAVA_SPI_DIR}"
+ for spi_module in jni-spi jni-bootstrap; do
+
spi_jar="${DORIS_HOME}/fe/be-java-extensions/${spi_module}/target/doris-${spi_module}.jar"
+ # Louder than the plugin loop below, not quieter: without these
two jars there is no
+ # loader at all, so every Java feature fails at runtime with a
FindClass error that
+ # names none of this. They are also not affected by
BE_EXTENSION_IGNORE - see the
+ # module list far above.
+ if [[ ! -f "${spi_jar}" ]]; then
+ echo "Error: ${spi_module} produced no ${spi_jar}. It carries
the plugin SPI and the"
+ echo " loader that reads plugins/jni, so a BE without it
can load no Java"
+ echo " plugin at all."
+ exit 1
+ fi
+ echo "Copy Be shared layer ${spi_module} jar to ${BE_JAVA_SPI_DIR}"
+ cp "${spi_jar}" "${BE_JAVA_SPI_DIR}"
+ done
+
+ # Plugins, one directory each: the module jar plus the runtime closure
copy-dependencies put
+ # beside it. The directory name is what BE addresses the plugin by and
is deliberately not
+ # required to equal the module name - paimon-scanner will deploy as
"paimon" - so the mapping
+ # is spelled out rather than derived.
+ #
+ # ATTN: a module named here must also be in the maven module list far
above; adding it in one
+ # place only means deploying whatever the last build happened to leave
in target/, which looks
+ # like a successful build of the wrong thing.
+ BE_JAVA_PLUGINS_DIR="${DORIS_OUTPUT}/be/plugins/jni"
+ # ATTN: this rm reaches into plugins/, which is otherwise the
operator's tree - the drivers,
+ # configs and UDF jars they dropped there. It must name plugins/jni
and nothing above it;
+ # widening it by one path element wipes a running deployment's drop
points.
+ rm -rf "${BE_JAVA_PLUGINS_DIR}"
+ mkdir -p "${BE_JAVA_PLUGINS_DIR}"
+ plugin_modules=("java-writer:java-writer")
+ plugin_modules+=("jdbc-scanner:jdbc")
+ plugin_modules+=("iceberg-metadata-scanner:iceberg")
+ plugin_modules+=("max-compute-connector:max-compute")
+ plugin_modules+=("paimon-scanner:paimon")
+ plugin_modules+=("hadoop-hudi-scanner:hudi")
+ plugin_modules+=("trino-connector-scanner:trino-connector")
+ plugin_modules+=("java-udf:java-udf")
+
+ if [[ -n "${BE_EXTENSION_IGNORE}" ]]; then
+ IFS=',' read -r -a ignore_modules <<<"${BE_EXTENSION_IGNORE}"
+ kept_plugins=()
+ for plugin_entry in "${plugin_modules[@]}"; do
ignore=0
for ignore_module in "${ignore_modules[@]}"; do
- if [[ "${module}" == "${ignore_module}" ]]; then
+ if [[ "${plugin_entry%%:*}" == "${ignore_module// /}" ]];
then
ignore=1
break
fi
done
if [[ "${ignore}" -eq 0 ]]; then
- new_modules+=("${module}")
+ kept_plugins+=("${plugin_entry}")
fi
+ done
+ plugin_modules=("${kept_plugins[@]}")
+ fi
+
+ for plugin_entry in "${plugin_modules[@]}"; do
+ plugin_module="${plugin_entry%%:*}"
+ plugin_name="${plugin_entry##*:}"
+
plugin_target="${DORIS_HOME}/fe/be-java-extensions/${plugin_module}/target"
+ plugin_jar="${plugin_target}/${plugin_module}.jar"
+ if [[ ! -f "${plugin_jar}" ]]; then
+ echo "Error: ${plugin_module} produced no
${plugin_module}.jar. A plugin jar is named"
+ echo " after its module; deploying an empty plugin
directory would surface much"
+ echo " later as 'Java plugin ${plugin_name} failed to
load'."
+ exit 1
+ fi
+ echo "Copy Be plugin ${plugin_module} to
${BE_JAVA_PLUGINS_DIR}/${plugin_name}"
+ mkdir -p "${BE_JAVA_PLUGINS_DIR}/${plugin_name}"
+ cp "${plugin_jar}" "${BE_JAVA_PLUGINS_DIR}/${plugin_name}"
+ # Tested on the jars, not on the directory: target/lib is emptied
before
+ # copy-dependencies refills it, so an existing but empty directory
is reachable and the
+ # glob below would then expand to nothing and fail the whole build
under set -e.
+ if compgen -G "${plugin_target}/lib/*.jar" > /dev/null; then
+ cp "${plugin_target}/lib"/*.jar
"${BE_JAVA_PLUGINS_DIR}/${plugin_name}"
fi
done
- extensions_modules=("${new_modules[@]}")
- fi
- BE_JAVA_EXTENSIONS_DIR="${DORIS_OUTPUT}/be/lib/java_extensions/"
- rm -rf "${BE_JAVA_EXTENSIONS_DIR}"
- mkdir "${BE_JAVA_EXTENSIONS_DIR}"
- for extensions_module in "${extensions_modules[@]}"; do
-
module_jar="${DORIS_HOME}/fe/be-java-extensions/${extensions_module}/target/${extensions_module}-jar-with-dependencies.jar"
-
module_proj_jar="${DORIS_HOME}/fe/be-java-extensions/${extensions_module}/target/${extensions_module}-project.jar"
- mkdir "${BE_JAVA_EXTENSIONS_DIR}"/"${extensions_module}"
- echo "Copy Be Extensions ${extensions_module} jar to
${BE_JAVA_EXTENSIONS_DIR}/${extensions_module}"
- if [[ "${extensions_module}" == "${HADOOP_DEPS_NAME}" ]]; then
-
+ # The hadoop drop C++ libhdfs loads, and the JindoFS/JuiceFS drops the
same libhdfs resolves
+ # oss-hdfs:// and jfs:// through: none of the three is a plugin, so
libhdfs finds each by a
+ # fixed directory name on the system classpath rather than through a
plugin loader. Each is
+ # therefore wiped and deployed whole every build rather than merged
with whatever a previous
+ # build using the same output directory left behind - unwiped, a
version bump would leave two
+ # jar versions of the same filesystem side by side, and start_be.sh's
*.jar glob would put
+ # both of them on the classpath.
+ if [[ "${deploy_hadoop_deps}" -eq 1 ]]; then
Review Comment:
[P1] Remove Hadoop jars when `hadoop-deps` is explicitly ignored
The old unconditional wipe was moved inside this `deploy_hadoop_deps`
branch. Consequently, a normal build followed by `--be-extension-ignore
hadoop-deps` into the same `DORIS_OUTPUT` skips both the copy and the only
cleanup, so `be/lib/hadoop_hdfs` retains the first build's jars and
`start_be.sh` still loads them. Fresh and reused outputs now disagree about an
explicitly excluded module. Preserve this directory only for the
whole-extension-disabled case if needed, but remove it when `hadoop-deps`
itself is named in `BE_EXTENSION_IGNORE`.
##########
fe/fe-connector/fe-connector-spi/src/main/java/org/apache/doris/connector/spi/ConnectorMetadata.java:
##########
@@ -98,6 +98,46 @@ default Optional<ConnectorMvccPartitionView>
getMvccPartitionView(
return Optional.empty();
}
+ /**
+ * Whether {@link #listPartitions} reads the pin carried by a
snapshot-applied handle, so that a
+ * {@code FOR TIME/VERSION AS OF} query can be told which partitions
existed AT that pin.
+ *
+ * <p>Point-in-time time travel otherwise pins with EMPTY partition maps,
because a partition set
+ * listed at LATEST is the wrong universe for a past snapshot in both
directions: it hides a
+ * partition that has since been dropped (pruning it away loses rows) and
invents ones created
+ * after the pin. Empty is the safe answer — the generic scan node reads
it as scan-all and lets
+ * the connector's own predicate pushdown do the pruning — but it costs
the query its partition
+ * pruning and makes EXPLAIN report {@code partition=0/0} for a scan that
reads everything.</p>
+ *
+ * <p>A connector that answers true promises that {@code listPartitions}
on the handle returned by
+ * {@link #applySnapshot} enumerates exactly the partitions with data at
that pin. The generic
+ * model then pins the real partition set and both pruning and {@code
partition=N/M} become
+ * truthful. The default is false: a connector whose listing is
snapshot-blind keeps the empty pin,
+ * which is correct, just coarse.</p>
+ *
+ * <p><b>A SECOND PROMISE COMES WITH IT, and it is easy to miss:</b> the
AT-SNAPSHOT schema this
+ * connector returns from {@link #getTableSchema(ConnectorSession,
ConnectorTableHandle,
+ * ConnectorMvccSnapshot)} must declare, through {@code
ConnectorTableSchema.PARTITION_COLUMNS_KEY},
+ * partition columns that MATCH the values that pinned listing produces -
same count, in the same
+ * order, each parseable into the column's type. That schema is what types
the pinned partition
+ * items, because it is the schema this snapshot publishes; the latest
schema may partition the
+ * table differently.</p>
+ *
+ * <p>Breaking that promise degrades quietly rather than failing: each
mismatched partition is
+ * skipped inside a per-partition catch, the pinned item map ends up
shorter than the listed name
+ * set, and the table is reported UNPARTITIONED. Rows are still correct -
pruning and
+ * {@code partition=N/M} are what is lost. Two signals in the log: one
WARN per skipped
+ * partition, and - when EVERY partition was skipped, which is what a
schema that never matched
+ * produces as opposed to iceberg spec evolution - one aggregate WARN
naming both counts and the
+ * partition columns, from {@code PluginDrivenMvccExternalTable}. Note
+ * that a connector whose at-snapshot schema resolution can degrade to an
empty column list on
+ * error - hudi's {@code getSchemaFromMetaClient} swallows a failed
metadata read into one -
+ * produces exactly this shape from a transient fault.</p>
+ */
+ default boolean listsPartitionsAtSnapshot(ConnectorSession session,
ConnectorTableHandle handle) {
Review Comment:
[P1] Bump the connector SPI major for this method
This expands the frozen connector SPI, while `fe/fe-connector/pom.xml`
explicitly requires a major bump for any method addition and still publishes
7.0. Since `ApiVersionGate` compares only majors, older and newer, mutually
different surfaces both identify as compatible; a new plugin can be admitted by
an old FE even though this method is absent there. Bump
`connector.plugin.api.version` to 8.0 and update the surface/version assertions
and generated manifests/resources in the same change.
##########
fe/be-java-extensions/jni-bootstrap/src/main/java/org/apache/doris/jni/bootstrap/PluginRuntime.java:
##########
@@ -0,0 +1,657 @@
+// 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.jni.bootstrap;
+
+import org.apache.doris.jni.spi.DorisPlugin;
+import org.apache.doris.jni.spi.JniScannerFactory;
+import org.apache.doris.jni.spi.JniWriterFactory;
+import org.apache.doris.jni.spi.SpiVersion;
+import org.apache.doris.jni.spi.ThreadContextClassLoader;
+import org.apache.doris.jni.spi.UdfExecutorFactory;
+import org.apache.doris.jni.spi.utils.JniUtil;
+
+import java.io.IOException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.ServiceLoader;
+import java.util.TreeMap;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * Loads plugins from a directory and hands BE the objects it asks for.
+ *
+ * <p>Loading is lazy and happens at most once per plugin: the first request
for a plugin builds its
+ * classloader, runs {@link ServiceLoader} and indexes the factories, and
every later request reads
+ * the cached {@link PluginHandle} - including a cached failure. Nothing is
eagerly resolved beyond
+ * the plugin object and its factories, so a class missing from a plugin's
jars is reported when
+ * something touches it, with the name of the class, instead of taking the
process down at startup.
+ *
+ * <p>All plugin code runs with the plugin's classloader installed as the
thread context
+ * classloader. BE's threads have no meaningful one, and ServiceLoader and
most plugin libraries
+ * consult it.
+ */
+final class PluginRuntime {
+
+ private static final Logger LOG =
Logger.getLogger(PluginRuntime.class.getName());
+
+ private final Path pluginDir;
+ private final ClassLoader spiClassLoader;
+ private final ClassLoader hadoopConfResources;
+ private final Path fsDir;
+ private final ConcurrentHashMap<String, PluginHandle> plugins = new
ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Object> loadLocks = new
ConcurrentHashMap<>();
+
+ PluginRuntime(Path pluginDir, ClassLoader spiClassLoader) {
+ this(pluginDir, spiClassLoader, null, null);
+ }
+
+ PluginRuntime(Path pluginDir, ClassLoader spiClassLoader, Path
hadoopConfDir) {
+ this(pluginDir, spiClassLoader, hadoopConfDir, null);
+ }
+
+ /**
+ * @param hadoopConfDir directory whose files every plugin can read as
classpath resources, so
+ * that a hadoop {@code Configuration} built inside a
plugin finds
+ * {@code core-site.xml} and friends. Null when there
is none.
+ * @param fsDir directory of third-party hadoop {@code FileSystem}
jars every plugin
+ * may need; see {@link #sharedFilesystemJars()}.
Null when there is none.
+ */
+ PluginRuntime(Path pluginDir, ClassLoader spiClassLoader, Path
hadoopConfDir, Path fsDir) {
+ this.pluginDir = Objects.requireNonNull(pluginDir, "pluginDir");
+ this.spiClassLoader = Objects.requireNonNull(spiClassLoader,
"spiClassLoader");
+ this.hadoopConfResources = hadoopConfLoader(hadoopConfDir);
+ this.fsDir = fsDir;
+ }
+
+ /**
+ * Jars appended to EVERY plugin's classpath, after that plugin's own.
+ *
+ * <p>What lives here: third-party hadoop {@code FileSystem}
implementations that no plugin
+ * declares as a dependency because none of them is written against it -
JindoFS serves
+ * {@code oss://} and {@code oss-hdfs://}, JuiceFS serves {@code jfs://},
and hadoop reaches
+ * both by class name out of a {@code Configuration}. Both are opt-in
build flags
+ * ({@code DISABLE_BUILD_JINDOFS=OFF}, {@code DISABLE_BUILD_JUICEFS=OFF}),
so on a default
+ * build this directory is absent and this method returns nothing.
+ *
+ * <p>Why shared rather than bundled per plugin. Before the plugins were
isolated these jars
+ * sat on the system classpath, which every scanner could reach, so any
table format could read
+ * a table on any of those filesystems. A plugin classloader cannot reach
that classpath by
+ * design, and the alternative - copying the jars into each plugin
directory - does not scale:
+ * the JuiceFS Hadoop SDK is a 180 MB fat jar that carries jersey,
checkerframework and
+ * javax.ws.rs, which collide with about 1500 classes already in a
lake-format plugin. One
+ * directory read by every plugin costs one copy on disk and produces no
collisions to
+ * adjudicate.
+ *
+ * <p>APPENDED, never prepended: these fat jars carry stray copies of
third-party classes,
+ * hadoop's included, and a plugin's own hadoop must win. Counted on the
jars this build
+ * packages: jindo-sdk carries 19 hadoop classes (11 in {@code
org.apache.hadoop.fs}, 5 in
+ * {@code fs.impl}, 3 in {@code util}) and juicefs-hadoop carries 4, all in
+ * {@code org.apache.hadoop.security}. That is the same rule {@code
bin/start_be.sh} applies
+ * when it puts them after {@code lib/hadoop_hdfs} on the system classpath
for libhdfs.
+ *
+ * <p>ISOLATION IS PRESERVED: each plugin loads its own copy of these
classes in its own
Review Comment:
[P1] Avoid loading Jindo's native library in every plugin
These shared jars are resolved independently by every plugin classloader,
but the comment below and `build.sh` both acknowledge that `jindo-core` can
bind its native library to only one classloader and that the second plugin (or
plugin plus libhdfs) fails. In the supplied base the BE Jindo jars were on one
system classpath; this change makes the failure reachable for ordinary mixed
Paimon/Iceberg/Hudi access to an OSS warehouse. Provide one JVM owner/bridge
for the native component, or another layout that shares the binding, and cover
sequential access from two real plugin classloaders.
##########
be/src/format/transformer/vfile_format_transformer_factory.cpp:
##########
@@ -49,8 +49,23 @@ Status create_tvf_format_transformer(const TTVFTableSink&
tvf_sink, RuntimeState
if (tvf_sink.__isset.line_delimiter) {
writer_params["line_delimiter"] = tvf_sink.line_delimiter;
}
- result->reset(new VJniFormatTransformer(state, output_vexpr_ctxs,
tvf_sink.writer_class,
- std::move(writer_params)));
+ // writer_class names a plugin factory, not a Java class. A class name
stopped being able
+ // to identify a writer when plugins were isolated: a concrete writer
lives in its own
+ // plugin's classloader, which BE cannot search by name, so what is
addressable is the
+ // plugin directory under plugins/jni and the factory inside it.
+ const std::string& writer = tvf_sink.writer_class;
+ const size_t sep = writer.find(':');
+ if (sep == std::string::npos || sep == 0 || sep + 1 == writer.size()) {
Review Comment:
[P1] Preserve `writer_class` through rolling upgrades
This reinterprets existing thrift field 15 in place. A legacy class name
works on an old BE but is explicitly rejected here, while
`java-writer:local-file` works here but an old BE passes it to `FindClass` and
cannot resolve it. FE forwards one user value and has no scheduled-BE
capability branch, so no spelling can execute on every BE during a rolling
upgrade. Keep accepting/mapping the known legacy class name until the
compatibility window closes, or introduce a versioned/capability-negotiated
field.
##########
be/CMakeLists.txt:
##########
@@ -686,29 +685,21 @@ set(COMMON_THIRDPARTY
${COMMON_THIRDPARTY}
)
-if ((ARCH_AMD64 OR ARCH_AARCH64) AND OS_LINUX)
- add_library(hadoop_hdfs STATIC IMPORTED)
- set_target_properties(hadoop_hdfs PROPERTIES IMPORTED_LOCATION
${THIRDPARTY_DIR}/lib/hadoop_hdfs_3_4/native/libhdfs.a)
+add_library(hadoop_hdfs STATIC IMPORTED)
Review Comment:
[P1] Keep the macOS no-JVM fallback usable
This removes libhdfs3 and makes Java libhdfs unconditional, but `build.sh`
still treats a missing or wrong-architecture macOS `libjvm.dylib` as a
supported fallback and writes `enable_java_support=false`. Every new HDFS
connection now calls `ensure_jvm()`, which rejects exactly that config, so the
packaged fallback can no longer access HDFS (the new unit test confirms the
refusal). Either retain a non-JVM HDFS implementation for this build or make a
compatible JDK/libjvm a hard macOS requirement and remove the fallback.
##########
fe/be-java-extensions/plugin-toolkit/src/main/java/org/apache/doris/jni/toolkit/jdbc/JdbcDriverUtils.java:
##########
@@ -0,0 +1,373 @@
+// 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.jni.toolkit.jdbc;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.net.URLConnection;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Loads a user-supplied JDBC driver jar into a classloader of its own.
+ *
+ * <h2>Why the driver gets its own classloader</h2>
+ *
+ * <p>The jar is named by the catalog and downloaded at query time, so it
cannot be part of any
+ * plugin's build. Giving it a child of the plugin's classloader means the
driver can see the JDBC
+ * API and the plugin's classes, the plugin cannot accidentally compile
against driver internals,
+ * and two catalogs pointing at different drivers stay independent.
+ *
+ * <h2>Why the classloader is cached</h2>
+ *
+ * <p>One per driver jar, for the life of the process. Building a fresh one
per scan looks harmless
+ * but is not: connection pools outlive a single scan, so a pooled connection
keeps the driver
+ * classes from the loader that created it while the next scan loads a second
copy of the same
+ * classes. Everything then works until something compares two driver types
and finds them
+ * unrelated. Caching also bounds a leak that is otherwise proportional to
query count - a
+ * classloader holding a jar's worth of classes is not cheap.
+ *
+ * <p>The cache is keyed by jar <em>and</em> parent, because "the same jar
under two plugins" is two
+ * different driver worlds. Note that a driver jar replaced in place at the
same URL is not picked
+ * up on its own; {@link #checksumVerifier}, which the JDBC scanner, writer
and connection tester
+ * all pass, is what turns that from a silent stale read into an error - and,
once the operator
+ * states the jar's new checksum, into a reload: an expectation that differs
from the one the
+ * cached loader was verified under discards it, because otherwise the check
would report success
+ * against bytes the process is not running.
+ *
+ * <p>Whether a jar has been checked is remembered SEPARATELY from the
classloader, keyed by jar and
+ * expected checksum rather than by jar and parent. Folding the two together
looks equivalent and is
+ * not: the connection tester runs first, under the same parent as the
scanner, so once it had built
+ * the loader every later scan took the cache's early return and never reached
its own verifier -
+ * one un-checksummed CREATE CATALOG disabled the check for that jar for the
life of the process.
+ */
+public final class JdbcDriverUtils {
+
+ /** Same 10s the executor this replaced used, for both connect and read. */
+ private static final int CHECKSUM_TIMEOUT_MS = 10000;
+
+ private static final ConcurrentHashMap<DriverKey, ClassLoader>
DRIVER_CLASS_LOADERS =
+ new ConcurrentHashMap<>();
+
+ /** One lock per driver jar, so that loading two different drivers does
not serialize. */
+ private static final ConcurrentHashMap<DriverKey, Object> LOAD_LOCKS = new
ConcurrentHashMap<>();
+
+ /**
+ * Jars already checked, as "<url>\0<expectation>". Not keyed by parent:
the bytes behind a URL
+ * are the same bytes whichever plugin asked, so one read answers for all
of them - which is the
+ * whole reason this is remembered at all. Two catalogs naming the same
URL with DIFFERENT
+ * checksums are two entries and both get checked; exactly one of them can
pass.
+ */
+ private static final Set<String> VERIFIED = ConcurrentHashMap.newKeySet();
+
+ /**
+ * Per driver jar URL, the expectation the classloaders cached for it were
built under.
+ *
+ * <p>This is what connects the two caches above, which are otherwise
deliberately independent.
+ * A new expectation for a URL that already has a loader means the
operator replaced the jar in
+ * place and told Doris its new checksum; the bytes just verified are then
NOT the bytes the
+ * cached loader was built from. Without this the check reports success
against the current jar
+ * while every query keeps using the old driver until BE restarts - a
verification that passes
+ * for a driver the process is not running.
+ *
+ * <p>An entry is written whenever a loader is created, NOT only when an
expectation was stated:
+ * a catalog defined without {@code jdbc_driver_checksum} produces no
expectation at all, and
+ * leaving it out of this map is what made the discard below unreachable
for exactly the case it
+ * exists for. The operator's sequence is "create the catalog, later
replace the jar and declare
+ * its checksum" - the first step has nothing to record and the second
then found no previous
+ * entry, read {@code null}, and skipped the discard. {@link #UNDECLARED}
is that first step's
+ * entry: it equals no checksum, so any checksum stated later differs from
it.
+ */
+ private static final ConcurrentHashMap<String, String> LOADED_UNDER = new
ConcurrentHashMap<>();
+
+ /**
+ * The {@link #LOADED_UNDER} value for a loader built without a stated
expectation. Not a valid
+ * checksum in any spelling - {@link #checksumVerifier} lower-cases hex -
so it can never be
+ * mistaken for one, and comparing it against a real checksum always says
"different".
+ */
+ private static final String UNDECLARED = "<no checksum declared>";
+
+ private JdbcDriverUtils() {
+ }
+
+ /**
+ * Checks a driver jar before it is loaded for the first time. Callers
that know what the jar
+ * should be - Doris ships an MD5 with the catalog definition - pass one;
the check then runs
+ * exactly once per jar, when its classloader is created, rather than on
every query.
+ */
+ public interface DriverJarVerifier {
+ /** Throws to reject the jar; the classloader is then not created or
cached. */
+ void verify(URL driverJar);
+
+ /**
+ * What this verifier expects of the jar, or null when it cannot say.
+ *
+ * <p>It is what lets "already checked" be remembered without holding
on to the verifier:
+ * two calls naming the same jar and the same expectation ask the same
question, so the
+ * second is skipped even when the classloader was built by somebody
else. A verifier that
+ * returns null keeps the older, weaker rule - it runs only when the
classloader is created
+ * - because nothing can tell two of them apart.
+ */
+ default String expectation() {
+ return null;
+ }
+ }
+
+ /**
+ * The verifier Doris ships with: the MD5 the catalog definition carries,
compared against the
+ * jar actually behind the driver URL.
+ *
+ * <p>Returns {@code null} - "do not verify" - when the expected checksum
is blank, which is
+ * what a catalog defined without one produces. Only what Doris was told
to expect is checked;
+ * this never invents an expectation of its own.
+ *
+ * <p>The read is what makes it worth caching: {@code driverClassLoader}
runs the verifier
+ * exactly once per jar and parent, when the classloader for it is
created, so a remote driver
+ * jar is downloaded for checksumming once per process and not once per
query.
+ */
+ public static DriverJarVerifier checksumVerifier(String expectedChecksum) {
+ if (expectedChecksum == null || expectedChecksum.trim().isEmpty()) {
+ return null;
+ }
+ String expected = expectedChecksum.trim();
+ return new DriverJarVerifier() {
+ @Override
+ public void verify(URL driverJar) {
+ String actual = md5Of(driverJar);
+ if (!expected.equalsIgnoreCase(actual)) {
+ throw new IllegalStateException("Checksum mismatch for
JDBC driver " + driverJar
+ + ": the catalog expects " + expected + " but the
jar is " + actual
+ + ". The driver jar behind this URL is not the one
the catalog was defined"
+ + " with; replace the jar or redefine the catalog
with the new checksum");
+ }
+ }
+
+ @Override
+ public String expectation() {
+ // Lower-cased because the comparison above is
case-insensitive: the same MD5 in two
+ // spellings is one question, and must not be asked twice.
+ return expected.toLowerCase(Locale.ROOT);
+ }
+ };
+ }
+
+ private static String md5Of(URL driverJar) {
+ try {
+ MessageDigest digest = MessageDigest.getInstance("MD5");
+ URLConnection connection = driverJar.openConnection();
+ connection.setConnectTimeout(CHECKSUM_TIMEOUT_MS);
+ connection.setReadTimeout(CHECKSUM_TIMEOUT_MS);
+ try (InputStream in = connection.getInputStream()) {
+ byte[] buffer = new byte[8192];
+ int read;
+ while ((read = in.read(buffer)) != -1) {
+ digest.update(buffer, 0, read);
+ }
+ }
+ StringBuilder hex = new StringBuilder(32);
+ for (byte b : digest.digest()) {
+ hex.append(Character.forDigit((b >> 4) & 0xF,
16)).append(Character.forDigit(b & 0xF, 16));
+ }
+ return hex.toString();
+ } catch (IOException | NoSuchAlgorithmException e) {
+ throw new IllegalStateException("Cannot checksum the JDBC driver
at " + driverJar
+ + ": " + e.getMessage(), e);
+ }
+ }
+
+ /** The classloader for one driver jar, creating it on first use. */
+ public static ClassLoader driverClassLoader(String driverUrl, ClassLoader
parent) {
+ return driverClassLoader(driverUrl, parent, null);
+ }
+
+ /**
+ * @param driverUrl where the driver jar lives, as a URL
+ * @param parent the classloader of the code that will use the driver
+ * @param verifier optional; runs once, before the classloader for this
jar exists
+ */
+ public static ClassLoader driverClassLoader(String driverUrl, ClassLoader
parent,
+ DriverJarVerifier verifier) {
+ DriverKey key = new DriverKey(toUrl(driverUrl), parent);
+ ClassLoader cached = DRIVER_CLASS_LOADERS.get(key);
+ if (cached != null) {
+ // Before the early return, not after the cache miss: a cached
loader says nothing about
+ // whether THIS caller's expectation was ever checked against the
jar.
+ verifyOnce(key.driverUrl, verifier, false);
+ // Re-read rather than return `cached`: verifyOnce discards the
loaders for this jar when
+ // the expectation it just checked is not the one they were built
under, and returning
+ // the loader read a moment ago would hand back exactly the stale
driver that discovery
+ // was for. A miss here falls through and builds a fresh loader
from the new bytes.
+ ClassLoader stillCached = DRIVER_CLASS_LOADERS.get(key);
+ if (stillCached != null) {
+ return stillCached;
+ }
+ }
+ // Per-key lock plus a second look, rather than computeIfAbsent:
verifying a driver jar
+ // reads and checksums it and creating the loader opens it, and
neither may run while a
+ // ConcurrentHashMap bin lock is held - two catalogs whose keys share
a bin would then
+ // serialize on each other's jar download, and a nested load would be
a recursive update.
+ // What computeIfAbsent bought is kept: exactly one classloader per
jar is ever published.
+ synchronized (LOAD_LOCKS.computeIfAbsent(key, entry -> new Object())) {
+ ClassLoader loaded = DRIVER_CLASS_LOADERS.get(key);
+ // Before the loader exists, so that a jar that fails the check is
neither opened nor
+ // cached. Throwing here leaves nothing behind and the next
request checks again.
+ verifyOnce(key.driverUrl, verifier, loaded == null);
+ // Re-read, for the same reason the fast path above does and with
the same consequence
+ // when it is skipped: verifyOnce may have just discarded the
loaders for this jar, and
+ // `loaded` was read before that. Returning it would hand back the
stale driver that the
+ // discard was for. Reached only when the fast path missed and
this one hit, which is a
+ // publication race rather than the ordinary ALTER - that one is
served above.
+ loaded = DRIVER_CLASS_LOADERS.get(key);
+ if (loaded == null) {
+ loaded = URLClassLoader.newInstance(new URL[] {key.driverUrl},
key.parent);
+ DRIVER_CLASS_LOADERS.put(key, loaded);
+ // Recorded here rather than only in verifyOnce, so that a
loader built with no
+ // stated expectation is still attributable. putIfAbsent
because verifyOnce has
+ // already written the real expectation when there was one,
and that is the more
+ // specific answer.
+ LOADED_UNDER.putIfAbsent(key.driverUrl.toString(),
declaredExpectation(verifier));
+ }
+ return loaded;
+ }
+ }
+
+ /**
+ * Runs the verifier unless this exact question - this jar, this
expectation - was already
+ * answered in this process.
+ *
+ * @param loaderIsNew whether the classloader is about to be created,
which is the only thing a
+ * verifier that cannot state its expectation can be
keyed on
+ */
+ /** What {@link #LOADED_UNDER} records for a loader built under {@code
verifier}. */
+ private static String declaredExpectation(DriverJarVerifier verifier) {
+ String expectation = verifier == null ? null : verifier.expectation();
+ return expectation == null ? UNDECLARED : expectation;
+ }
+
+ private static void verifyOnce(URL driverJar, DriverJarVerifier verifier,
boolean loaderIsNew) {
+ if (verifier == null) {
+ return;
+ }
+ String expectation = verifier.expectation();
+ if (expectation == null) {
+ if (loaderIsNew) {
+ verifier.verify(driverJar);
+ }
+ return;
+ }
+ String token = driverJar.toString() + '\0' + expectation;
+ if (VERIFIED.contains(token)) {
Review Comment:
[P1] Invalidate the loader after a checksum rollback
`VERIFIED` lives for the process lifetime. After checksums A and B have both
been used for this URL, rolling the jar back to A hits this early return before
`LOADED_UNDER` is changed from B or `dropLoaders` runs, so the cached B loader
is returned while the caller supplied expectation A. This also affects the
connection tester independently of the datasource-cache issue. Treat a token as
current only while `LOADED_UNDER` equals it, update/invalidate the current
generation atomically, and add an A-to-B-to-A test.
--
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]
