This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 7f35e7eedcc0 test(spark): guard against executor .hoodie access and
task payload growth on the read path (#20112)
7f35e7eedcc0 is described below
commit 7f35e7eedcc002a5a1dd5f68b530a0911dc74d90
Author: Y Ethan Guo <[email protected]>
AuthorDate: Tue Oct 6 23:07:45 2026 -0700
test(spark): guard against executor .hoodie access and task payload growth
on the read path (#20112)
* test(spark): guard against executor .hoodie access and task payload
growth on the read path
* Replace the meta folder access recorder with a general file system call
recorder
* Enable the partition listing rows now that #20080 is merged
* Move the guards that wait on open fixes out of this change and read every
query in one meta folder test now that #20080 is merged
* Guard what every read task deserializes with per-read-shape size budgets
and against per-file Hadoop default resource parsing
* Check the task size budget on version 6 COPY_ON_WRITE reads too
* Read tables with a history of 50 commits and check the task size budget
on COPY_ON_WRITE incremental and time travel reads
* Keep MERGE_ON_READ log files uncompacted with the history, forbid any
timeline per task, check that data skipping reads the column stats index, and
drop unused recorder methods
* Key the task size budgets on columnar and row reads measured on Spark 3.5
and Spark 4.0
* Set the task size budgets per Spark and Scala binary version at about
1.1x the sizes measured on each
* Prune files with data skipping in strict mode, time travel to the first
commit, and check the task size budget on every query of both table versions
and types
* Check MERGE_ON_READ time travel against the row read budget and that time
travel sees only the first commit
* Keep the once per executor scan state apart from per-task streams, forbid
timeline state per task, and record file system calls of cached file systems
and task-started threads
---
.../testutils/TaskDeserializationRecorder.java | 340 ++++++++++++
.../apache/hudi/testutils/SparkExecutorGuards.java | 515 ++++++++++++++++++
.../functional/TestSparkReadExecutorFootprint.java | 597 +++++++++++++++++++++
3 files changed, 1452 insertions(+)
diff --git
a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/TaskDeserializationRecorder.java
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/TaskDeserializationRecorder.java
new file mode 100644
index 000000000000..8b24d4753e76
--- /dev/null
+++
b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/testutils/TaskDeserializationRecorder.java
@@ -0,0 +1,340 @@
+/*
+ * 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.hudi.testutils;
+
+import java.io.ObjectInputFilter;
+import java.io.ObjectInputStream;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
+
+/**
+ * Records what engine tasks Java-deserialize, through a JVM-wide {@link
ObjectInputFilter}.
+ *
+ * <p>An engine ships each task its closure (for Spark, the task binary
holding the RDD and the
+ * function to run) as a Java-serialized stream that every task deserializes
again. Whatever a
+ * closure captures therefore costs deserialization CPU and memory per task,
which is why heavy
+ * driver-side objects (a table meta client, its timeline, a full Hadoop
configuration) must not be
+ * captured. A functional test uses this recorder to assert on the classes
tasks deserialize and on
+ * the size of the largest task stream.
+ *
+ * <p>Streams read under an exempt stack frame (for Spark, a broadcast
variable fetch, which is
+ * deserialized once per executor rather than once per task) are recorded
separately and do not
+ * count towards {@link Result#getMaxStreamBytes()}.
+ *
+ * <p>The filter never rejects anything: it always answers {@link
ObjectInputFilter.Status#UNDECIDED},
+ * so installing it does not change what deserializes. It is installed once
per JVM and stays
+ * installed; outside {@link #start}/{@link #stop} it only reads a volatile
field.
+ */
+public final class TaskDeserializationRecorder implements ObjectInputFilter {
+
+ private static final TaskDeserializationRecorder INSTANCE = new
TaskDeserializationRecorder();
+ private static final int MAX_STACK_FRAMES = 30;
+ private static final Set<String> CLASS_DESCRIPTOR_METHODS = new
HashSet<>(Arrays.asList(
+ "filterCheck", "readClassDesc", "readNonProxyDesc", "readProxyDesc",
"readClassDescriptor"));
+
+ private volatile Session session;
+
+ private TaskDeserializationRecorder() {
+ }
+
+ /**
+ * Installs the recorder as the JVM-wide serialization filter if it is not
installed yet.
+ *
+ * @throws IllegalStateException if another JVM-wide filter is already
installed; the JDK accepts
+ * only one per JVM, so this recorder cannot
be chained after it
+ */
+ public static synchronized TaskDeserializationRecorder install() {
+ ObjectInputFilter current = ObjectInputFilter.Config.getSerialFilter();
+ if (current == INSTANCE) {
+ return INSTANCE;
+ }
+ if (current != null) {
+ throw new IllegalStateException("A JVM-wide serialization filter is
already installed (" + current
+ + ", e.g. from -Djdk.serialFilter), and the JDK allows only one per
JVM, so "
+ + TaskDeserializationRecorder.class.getSimpleName() + " cannot
record task deserialization. "
+ + "Run this test without that filter.");
+ }
+ ObjectInputFilter.Config.setSerialFilter(INSTANCE);
+ return INSTANCE;
+ }
+
+ /**
+ * Starts recording, keeping what each task scope deserializes apart so that
{@link #stop(Predicate)}
+ * can keep only the scopes of the work under test, for example the stages
of one job.
+ *
+ * @param taskScope the scope of the task the current thread runs, or
null outside tasks
+ * @param exemptFrameKind the kind of deserialization that is not per task,
such as a broadcast
+ * variable fetch, that a stack frame of the given
class marks, or null
+ */
+ public synchronized void start(Supplier<String> taskScope, Function<String,
String> exemptFrameKind) {
+ if (session != null) {
+ throw new IllegalStateException("A recording is already in progress");
+ }
+ session = new Session(taskScope, exemptFrameKind);
+ }
+
+ /**
+ * Stops recording and returns what the task scopes accepted by {@code
keepScope} deserialized
+ * since {@link #start}.
+ */
+ public synchronized Result stop(Predicate<String> keepScope) {
+ Session current = session;
+ if (current == null) {
+ throw new IllegalStateException("No recording in progress");
+ }
+ session = null;
+ return new Result(current, keepScope);
+ }
+
+ @Override
+ public Status checkInput(FilterInfo filterInfo) {
+ Session current = session;
+ if (current == null) {
+ return Status.UNDECIDED;
+ }
+ try {
+ String scope = current.taskScope.get();
+ if (scope != null) {
+ ScopeRecord record = current.scopes.computeIfAbsent(scope, s -> new
ScopeRecord());
+ Class<?> clazz = filterInfo.serialClass();
+ String exemptKind = StackWalker.getInstance().walk(frames -> frames
+ .map(frame -> current.exemptFrameKind.apply(frame.getClassName()))
+ .filter(Objects::nonNull)
+ .findFirst()
+ .orElse(null));
+ // A Class object in the stream, such as the runtime class of a
ClassTag, is not an instance.
+ boolean instance = clazz != null && !isClassReference();
+ if (exemptKind != null) {
+ // A stream reports a class once, so this counts the exempt
deserializations reading the class.
+ if (instance) {
+ record.exemptDeserializations.computeIfAbsent(exemptKind, kind ->
new ConcurrentHashMap<>())
+ .computeIfAbsent(clazz, c -> new
AtomicLong()).incrementAndGet();
+ }
+ } else {
+ if (instance && record.classes.add(clazz)) {
+ record.firstStacks.putIfAbsent(clazz, stackOfCaller());
+ }
+ record.maxStreamBytes.accumulateAndGet(filterInfo.streamBytes(),
Math::max);
+ }
+ }
+ } catch (Throwable t) {
+ // Recording must never interfere with deserialization.
+ current.errors.incrementAndGet();
+ }
+ return Status.UNDECIDED;
+ }
+
+ /**
+ * Whether the class descriptor being checked belongs to a {@code Class}
object in the stream, for
+ * example the runtime class of a {@code ClassTag}, rather than to an
instance being deserialized.
+ */
+ private static boolean isClassReference() {
+ return StackWalker.getInstance().walk(frames -> frames
+ .filter(frame ->
frame.getClassName().equals(ObjectInputStream.class.getName()))
+ .map(StackWalker.StackFrame::getMethodName)
+ .filter(method -> !CLASS_DESCRIPTOR_METHODS.contains(method))
+ .findFirst()
+ .map("readClass"::equals)
+ .orElse(false));
+ }
+
+ private static String stackOfCaller() {
+ return StackWalker.getInstance().walk(frames -> frames
+ .skip(1)
+ .filter(frame ->
!frame.getClassName().startsWith("java.io.ObjectInputStream")
+ &&
!frame.getClassName().equals(TaskDeserializationRecorder.class.getName()))
+ .limit(MAX_STACK_FRAMES)
+ .map(frame -> "\tat " + frame)
+ .collect(Collectors.joining("\n")));
+ }
+
+ /**
+ * Whether {@code clazz} is the class named {@code className} or extends or
implements it.
+ */
+ public static boolean isSubtypeOf(Class<?> clazz, String className) {
+ if (clazz == null) {
+ return false;
+ }
+ if (clazz.getName().equals(className) ||
isSubtypeOf(clazz.getSuperclass(), className)) {
+ return true;
+ }
+ for (Class<?> iface : clazz.getInterfaces()) {
+ if (isSubtypeOf(iface, className)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static final class Session {
+ private final Supplier<String> taskScope;
+ private final Function<String, String> exemptFrameKind;
+ private final Map<String, ScopeRecord> scopes = new ConcurrentHashMap<>();
+ private final AtomicLong errors = new AtomicLong();
+
+ private Session(Supplier<String> taskScope, Function<String, String>
exemptFrameKind) {
+ this.taskScope = taskScope;
+ this.exemptFrameKind = exemptFrameKind;
+ }
+ }
+
+ private static final class ScopeRecord {
+ private final Set<Class<?>> classes = ConcurrentHashMap.newKeySet();
+ private final Map<Class<?>, String> firstStacks = new
ConcurrentHashMap<>();
+ private final Map<String, Map<Class<?>, AtomicLong>>
exemptDeserializations = new ConcurrentHashMap<>();
+ private final AtomicLong maxStreamBytes = new AtomicLong();
+ }
+
+ /**
+ * What tasks deserialized during one recording.
+ */
+ public static final class Result {
+ private final Set<Class<?>> classes;
+ private final Map<Class<?>, String> firstStacks;
+ private final Map<String, Map<Class<?>, Long>> exemptDeserializations;
+ private final Set<String> keptScopes;
+ private final Set<String> ignoredScopes;
+ private final long maxStreamBytes;
+ private final long errors;
+
+ private Result(Session session, Predicate<String> keepScope) {
+ Set<Class<?>> keptClasses = new HashSet<>();
+ Map<Class<?>, String> keptFirstStacks = new HashMap<>();
+ Map<String, Map<Class<?>, Long>> keptExemptDeserializations = new
HashMap<>();
+ Set<String> kept = new TreeSet<>();
+ Set<String> ignored = new TreeSet<>();
+ long maxBytes = 0;
+ for (Map.Entry<String, ScopeRecord> entry : session.scopes.entrySet()) {
+ if (keepScope.test(entry.getKey())) {
+ kept.add(entry.getKey());
+ keptClasses.addAll(entry.getValue().classes);
+ entry.getValue().firstStacks.forEach(keptFirstStacks::putIfAbsent);
+ entry.getValue().exemptDeserializations.forEach((kind, counts) ->
counts.forEach((clazz, count) ->
+ keptExemptDeserializations.computeIfAbsent(kind, k -> new
HashMap<>()).merge(clazz, count.get(), Long::sum)));
+ maxBytes = Math.max(maxBytes, entry.getValue().maxStreamBytes.get());
+ } else {
+ ignored.add(entry.getKey());
+ }
+ }
+ this.classes = Collections.unmodifiableSet(keptClasses);
+ this.firstStacks = Collections.unmodifiableMap(keptFirstStacks);
+ this.exemptDeserializations =
Collections.unmodifiableMap(keptExemptDeserializations);
+ this.keptScopes = Collections.unmodifiableSet(kept);
+ this.ignoredScopes = Collections.unmodifiableSet(ignored);
+ this.maxStreamBytes = maxBytes;
+ this.errors = session.errors.get();
+ }
+
+ /**
+ * How many deserializations of the given exempt kind read an instance of
the class named
+ * {@code className}.
+ */
+ public long getExemptDeserializations(String kind, String className) {
+ return exemptDeserializations.getOrDefault(kind,
Collections.emptyMap()).entrySet().stream()
+ .filter(entry -> entry.getKey().getName().equals(className))
+ .mapToLong(Map.Entry::getValue)
+ .sum();
+ }
+
+ /**
+ * Names of the classes that are or extend {@code className} and that
deserializations of the
+ * given exempt kind read an instance of.
+ */
+ public List<String> exemptNamesOfSubtypesOf(String kind, String className)
{
+ return exemptDeserializations.getOrDefault(kind,
Collections.emptyMap()).keySet().stream()
+ .filter(clazz -> isSubtypeOf(clazz, className))
+ .map(Class::getName)
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * Where a task first deserialized the class named {@code className}
outside exempt frames, or
+ * null if no task did.
+ */
+ public String getFirstStack(String className) {
+ return firstStacks.entrySet().stream()
+ .filter(entry -> entry.getKey().getName().equals(className))
+ .map(Map.Entry::getValue)
+ .findFirst().orElse(null);
+ }
+
+ /**
+ * The task scopes whose deserialization this result holds.
+ */
+ public Set<String> getKeptScopes() {
+ return keptScopes;
+ }
+
+ /**
+ * The task scopes recorded but left out of this result.
+ */
+ public Set<String> getIgnoredScopes() {
+ return ignoredScopes;
+ }
+
+ /**
+ * Classes deserialized by tasks outside exempt frames.
+ */
+ public Set<Class<?>> getClasses() {
+ return classes;
+ }
+
+ /**
+ * The largest byte count any task stream outside exempt frames had
consumed when the filter
+ * was consulted, a close lower bound of the largest per-task stream.
+ */
+ public long getMaxStreamBytes() {
+ return maxStreamBytes;
+ }
+
+ /**
+ * How many filter callbacks failed to record; non-zero means the result
is incomplete.
+ */
+ public long getErrors() {
+ return errors;
+ }
+
+ /**
+ * Names of the classes deserialized outside exempt frames that are {@code
className} or extend
+ * or implement it. Matching by name lets a caller check for classes it
cannot compile against.
+ */
+ public List<String> namesOfSubtypesOf(String className) {
+ return classes.stream()
+ .filter(clazz -> isSubtypeOf(clazz, className))
+ .map(Class::getName)
+ .sorted()
+ .collect(Collectors.toList());
+ }
+ }
+}
diff --git
a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/SparkExecutorGuards.java
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/SparkExecutorGuards.java
new file mode 100644
index 000000000000..1caf74b3411c
--- /dev/null
+++
b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/testutils/SparkExecutorGuards.java
@@ -0,0 +1,515 @@
+/*
+ * 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.hudi.testutils;
+
+import org.apache.hudi.hadoop.fs.RecordingLocalFileSystem;
+import org.apache.hudi.hadoop.fs.RecordingLocalFileSystem.Call;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.spark.SparkContext;
+import org.apache.spark.SparkEnv;
+import org.apache.spark.TaskContext;
+import org.apache.spark.TaskFailedReason;
+import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.api.plugin.DriverPlugin;
+import org.apache.spark.api.plugin.ExecutorPlugin;
+import org.apache.spark.api.plugin.SparkPlugin;
+import org.apache.spark.rdd.RDD;
+import org.apache.spark.sql.Dataset;
+
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.io.OutputStream;
+import java.io.UncheckedIOException;
+import java.net.URI;
+import java.net.URL;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Queue;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BooleanSupplier;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import scala.reflect.ClassTag$;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Guards for what Spark tasks do on executors: that they do not touch a
table's {@code .hoodie}
+ * folder, that they do not deserialize heavy driver-side objects with the
task closure, and that
+ * they do not parse the Hadoop default resources for every file they read.
+ *
+ * <p>The guards need the work under test to run as Spark tasks in this JVM,
which a local session
+ * does.
+ */
+public final class SparkExecutorGuards {
+
+ /**
+ * The attempt id of the Spark task that is running on, or that started, the
current thread, set
+ * by {@link TaskStartHookPlugin}. Threads a task creates inherit it.
+ */
+ private static final InheritableThreadLocal<Long> TASK_ATTEMPT = new
InheritableThreadLocal<>();
+ private static final Set<Long> RUNNING_TASK_ATTEMPTS =
ConcurrentHashMap.newKeySet();
+
+ /**
+ * Whether the current thread runs a Spark task, or was started by a Spark
task that is still
+ * running (needs {@link TaskStartHookPlugin}). The thread name also covers
the part of a task run
+ * that precedes its {@link TaskContext}, namely deserializing the task and
its partition.
+ */
+ public static final BooleanSupplier IN_SPARK_TASK = () -> {
+ if (TaskContext.get() != null ||
Thread.currentThread().getName().startsWith("Executor task launch worker")) {
+ return true;
+ }
+ Long startedBy = TASK_ATTEMPT.get();
+ return startedBy != null && RUNNING_TASK_ATTEMPTS.contains(startedBy);
+ };
+
+ /**
+ * The kind of deserialization a broadcast variable fetch is, which happens
once per executor rather
+ * than once per task.
+ */
+ public static final String BROADCAST_FETCH = "broadcast fetch";
+
+ /**
+ * The kind of deserialization the value of the file group reader's broadcast
+ * {@code JavaSerializedValue} is: the scan state, deserialized once per JVM
instance of the holder.
+ */
+ public static final String SCAN_STATE = "scan state";
+
+ private static final String BROADCAST_CLASS =
"org.apache.spark.broadcast.TorrentBroadcast";
+ private static final String SCAN_STATE_HOLDER_CLASS =
"org.apache.spark.sql.execution.datasources.parquet.JavaSerializedValue";
+
+ /**
+ * The {@code spark.plugins} value that enables {@link #setTaskStartHook}.
+ */
+ public static final String TASK_START_HOOK_PLUGIN =
TaskStartHookPlugin.class.getName();
+
+ private static final String SPARK_JOB_GROUP_ID = "spark.jobGroup.id";
+ private static final String SPARK_JOB_DESCRIPTION = "spark.job.description";
+ private static final String SPARK_JOB_INTERRUPT_ON_CANCEL =
"spark.job.interruptOnCancel";
+ private static final Pattern TASK_THREAD_STAGE = Pattern.compile("^Executor
task launch worker .* in stage (\\d+)\\.");
+
+ private static final int MAX_WRITTEN_BEFORE = 12;
+ private static final int MAX_LOAD_SITES = 3;
+ private static final int MAX_LOAD_SITE_FRAMES = 40;
+
+ private static final URI LOCAL_FILE_SYSTEM = URI.create("file:///");
+
+ private static volatile Runnable taskStartHook = () -> { };
+
+ private SparkExecutorGuards() {
+ }
+
+ /**
+ * Routes the {@code file} scheme through {@link RecordingLocalFileSystem}
for every reader created
+ * from {@code hadoopConf} (use the Spark context's Hadoop configuration),
with Spark tasks as the
+ * recording scope. Register it before writing the table too, so that all
files are written and read
+ * by the same file system implementation.
+ *
+ * <p>The recording file system also replaces the cached {@code file} file
system, so that a reader
+ * that resolves its file system from another configuration, such as a new
one, is recorded too.
+ */
+ public static void enableFileSystemCallRecording(Configuration hadoopConf) {
+ RecordingLocalFileSystem.setScope(IN_SPARK_TASK);
+ RecordingLocalFileSystem.register(hadoopConf);
+ try {
+ FileSystem cached = FileSystem.get(LOCAL_FILE_SYSTEM, new
Configuration());
+ if (!(cached instanceof RecordingLocalFileSystem)) {
+ // Closing removes it from the cache, so that the next lookup caches
the recording file system.
+ cached.close();
+ Configuration recording = new Configuration();
+ recording.set(RecordingLocalFileSystem.FILE_IMPL_KEY,
RecordingLocalFileSystem.class.getName());
+ FileSystem.get(LOCAL_FILE_SYSTEM, recording);
+ }
+ assertTrue(FileSystem.get(LOCAL_FILE_SYSTEM, new Configuration())
instanceof RecordingLocalFileSystem,
+ "The cached file system of the file scheme must be the recording
one");
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ /**
+ * Reverts {@link #enableFileSystemCallRecording}.
+ */
+ public static void disableFileSystemCallRecording(Configuration hadoopConf) {
+ RecordingLocalFileSystem.unregister(hadoopConf);
+ RecordingLocalFileSystem.setScope(() -> false);
+ try {
+ FileSystem cached = FileSystem.get(LOCAL_FILE_SYSTEM, new
Configuration());
+ if (cached instanceof RecordingLocalFileSystem) {
+ cached.close();
+ }
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ /**
+ * Runs {@code action} and fails if any Spark task accessed a path under
{@code .hoodie} while it
+ * ran. Also fails if no Spark task call was recorded at all, which means
the tasks do not use the
+ * recording file system and the guard would pass vacuously.
+ */
+ public static <T> T assertNoExecutorMetaFolderAccess(String description,
Supplier<T> action) {
+ RecordingLocalFileSystem.reset();
+ T result = action.get();
+ assertTrue(RecordingLocalFileSystem.count(Call.inScope()) > 0,
+ description + ": no file system call of a Spark task was recorded, so
the guard is not live. Register "
+ + RecordingLocalFileSystem.class.getSimpleName()
+ + " on the Hadoop configuration the read uses
(enableFileSystemCallRecording).");
+ Predicate<Call> executorMetaFolderAccess =
Call.inScope().and(Call.underMetaFolder());
+ if (RecordingLocalFileSystem.count(executorMetaFolderAccess) > 0) {
+ fail(description + ": Spark tasks accessed the .hoodie folder. Table
config, timeline and schema"
+ + " history must be resolved once on the driver and handed to the
tasks; every access below"
+ + " is repeated per task (per file group) on the executors.\n"
+ + RecordingLocalFileSystem.describe(executorMetaFolderAccess));
+ }
+ return result;
+ }
+
+ /**
+ * Runs {@code action} while recording the classes Spark tasks
Java-deserialize, keeping only the
+ * tasks of the jobs {@code action} starts. Tasks of jobs that run
concurrently, such as file
+ * listing jobs started by planning in another thread, are left out.
+ */
+ public static TaskDeserializationRecorder.Result
recordTaskDeserialization(SparkContext sparkContext, Runnable action) {
+ String jobGroup = "task-deserialization-" + UUID.randomUUID();
+ Set<String> scopesOfJobGroup = ConcurrentHashMap.newKeySet();
+ Supplier<String> stageScope = () -> {
+ TaskContext taskContext = TaskContext.get();
+ if (taskContext != null) {
+ String scope = "stage " + taskContext.stageId();
+ if (jobGroup.equals(taskContext.getLocalProperty(SPARK_JOB_GROUP_ID)))
{
+ scopesOfJobGroup.add(scope);
+ }
+ return scope;
+ }
+ Matcher matcher =
TASK_THREAD_STAGE.matcher(Thread.currentThread().getName());
+ return matcher.find() ? "stage " + matcher.group(1) : null;
+ };
+ TaskDeserializationRecorder recorder =
TaskDeserializationRecorder.install();
+ String previousJobGroup =
sparkContext.getLocalProperty(SPARK_JOB_GROUP_ID);
+ String previousDescription =
sparkContext.getLocalProperty(SPARK_JOB_DESCRIPTION);
+ String previousInterrupt =
sparkContext.getLocalProperty(SPARK_JOB_INTERRUPT_ON_CANCEL);
+ sparkContext.setJobGroup(jobGroup, "task deserialization recording",
false);
+ recorder.start(stageScope, SparkExecutorGuards::oncePerExecutorKind);
+ TaskDeserializationRecorder.Result result;
+ try {
+ action.run();
+ } finally {
+ result = recorder.stop(scopesOfJobGroup::contains);
+ sparkContext.setLocalProperty(SPARK_JOB_GROUP_ID, previousJobGroup);
+ sparkContext.setLocalProperty(SPARK_JOB_DESCRIPTION,
previousDescription);
+ sparkContext.setLocalProperty(SPARK_JOB_INTERRUPT_ON_CANCEL,
previousInterrupt);
+ }
+ return result;
+ }
+
+ private static String oncePerExecutorKind(String frameClass) {
+ if (frameClass.startsWith(BROADCAST_CLASS)) {
+ return BROADCAST_FETCH;
+ }
+ return frameClass.startsWith(SCAN_STATE_HOLDER_CLASS) ? SCAN_STATE : null;
+ }
+
+ /**
+ * Sets code to run on the task thread at the start of every Spark task,
after the task and its
+ * partition are deserialized and before the task runs. It takes effect only
in a session whose {@code spark.plugins} includes
+ * {@link TaskStartHookPlugin}, see {@link #TASK_START_HOOK_PLUGIN}.
+ */
+ public static void setTaskStartHook(Runnable hook) {
+ taskStartHook = hook;
+ }
+
+ /**
+ * A Spark plugin that runs the hook set by {@link #setTaskStartHook} when a
task starts and tracks
+ * the running tasks for the threads they start, see {@link #IN_SPARK_TASK}.
+ */
+ public static final class TaskStartHookPlugin implements SparkPlugin {
+ @Override
+ public DriverPlugin driverPlugin() {
+ return null;
+ }
+
+ @Override
+ public ExecutorPlugin executorPlugin() {
+ return new ExecutorPlugin() {
+ @Override
+ public void onTaskStart() {
+ TaskContext context = TaskContext.get();
+ if (context != null) {
+ RUNNING_TASK_ATTEMPTS.add(context.taskAttemptId());
+ TASK_ATTEMPT.set(context.taskAttemptId());
+ }
+ taskStartHook.run();
+ }
+
+ @Override
+ public void onTaskSucceeded() {
+ taskEnded();
+ }
+
+ @Override
+ public void onTaskFailed(TaskFailedReason failureReason) {
+ taskEnded();
+ }
+
+ // Spark may unset the task context before these callbacks, so the
attempt comes from the thread.
+ private void taskEnded() {
+ Long attempt = TASK_ATTEMPT.get();
+ if (attempt != null) {
+ RUNNING_TASK_ATTEMPTS.remove(attempt);
+ }
+ TASK_ATTEMPT.remove();
+ }
+ };
+ }
+ }
+
+ /**
+ * Runs {@code action} and fails if Spark tasks parsed the Hadoop default
resources more than
+ * {@code maxLoads} times, which every {@link Configuration} created with
defaults does on its first
+ * read. A task must reuse the configuration it was given instead, since the
parse is repeated for
+ * every configuration created this way, typically once per file read.
+ *
+ * <p>Each task's context class loader is wrapped, once the task starts
running, to count the lookups
+ * of {@code core-default.xml}. A new configuration takes the context class
loader of the thread that
+ * creates it, so the count covers configurations the task creates while it
runs; a copy keeps the
+ * class loader of the configuration it copies. Needs a session with {@link
#TASK_START_HOOK_PLUGIN}.
+ * After {@code action}, a single task that creates a configuration checks
that the count is live.
+ */
+ public static <T> T assertTaskHadoopDefaultResourceLoadsAtMost(String
description, SparkContext sparkContext,
+ int maxLoads,
Supplier<T> action) {
+ AtomicInteger tasks = new AtomicInteger();
+ AtomicInteger loads = new AtomicInteger();
+ Queue<Throwable> loadSites = new ConcurrentLinkedQueue<>();
+ T result;
+ setTaskStartHook(() -> {
+ tasks.incrementAndGet();
+ Thread thread = Thread.currentThread();
+ thread.setContextClassLoader(
+ new
DefaultResourceCountingClassLoader(thread.getContextClassLoader(), loads,
loadSites));
+ });
+ try {
+ result = action.get();
+ int actionTasks = tasks.get();
+ int actionLoads = loads.get();
+ List<Throwable> actionLoadSites = new ArrayList<>(loadSites);
+
JavaSparkContext.fromSparkContext(sparkContext).parallelize(Collections.singletonList(1),
1)
+ .foreach(i -> new Configuration().get("fs.defaultFS"));
+ assertTrue(actionTasks > 0 && loads.get() > actionLoads,
+ description + ": the Hadoop default resource count is not live
(tasks of the action: " + actionTasks
+ + "); run the session with " + TASK_START_HOOK_PLUGIN + " in
spark.plugins");
+ if (actionLoads > maxLoads) {
+ fail(description + ": Spark tasks parsed the Hadoop default resources
" + actionLoads + " times in "
+ + actionTasks + " tasks, more than " + maxLoads + ". A reader must
use the configuration of the task,"
+ + " not create one per file, which parses core-default.xml and
core-site.xml each time. First parses:\n"
+ +
actionLoadSites.stream().map(SparkExecutorGuards::describeLoadSite)
+ .collect(Collectors.joining("\n")));
+ }
+ } finally {
+ setTaskStartHook(() -> { });
+ }
+ return result;
+ }
+
+ private static String describeLoadSite(Throwable site) {
+ return Arrays.stream(site.getStackTrace())
+ .skip(1)
+ .limit(MAX_LOAD_SITE_FRAMES)
+ .map(frame -> " at " + frame)
+ .collect(Collectors.joining("\n"));
+ }
+
+ /**
+ * Counts lookups of {@code core-default.xml}, which a {@link Configuration}
makes once per parse of
+ * its default resources, and keeps the stack of the first few.
+ */
+ private static final class DefaultResourceCountingClassLoader extends
ClassLoader {
+ private final AtomicInteger loads;
+ private final Queue<Throwable> loadSites;
+
+ private DefaultResourceCountingClassLoader(ClassLoader parent,
AtomicInteger loads, Queue<Throwable> loadSites) {
+ super(parent);
+ this.loads = loads;
+ this.loadSites = loadSites;
+ }
+
+ @Override
+ public URL getResource(String name) {
+ if ("core-default.xml".equals(name) && loads.incrementAndGet() <=
MAX_LOAD_SITES) {
+ loadSites.add(new Throwable());
+ }
+ return super.getResource(name);
+ }
+ }
+
+ /**
+ * The task binary Spark ships to every task that reads {@code df}: the RDD
lineage of its physical
+ * plan, Java-serialized as the DAG scheduler does. Call it after the plan
ran; it does not run a job.
+ */
+ public static TaskBinary inspectTaskBinary(Dataset<?> df) {
+ RDD<?> rdd = df.queryExecution().executedPlan().execute();
+ long bytes = SparkEnv.get().closureSerializer().newInstance()
+ .serialize(rdd, ClassTag$.MODULE$.apply(RDD.class)).remaining();
+ Map<Class<?>, List<String>> instances = new LinkedHashMap<>();
+ Deque<String> recentlyWritten = new ArrayDeque<>();
+ try (ObjectOutputStream out = new
ObjectOutputStream(OutputStream.nullOutputStream()) {
+ {
+ enableReplaceObject(true);
+ }
+
+ @Override
+ protected Object replaceObject(Object obj) {
+ instances.computeIfAbsent(obj.getClass(), c -> new
ArrayList<>(recentlyWritten));
+ recentlyWritten.addLast(obj.getClass().getName());
+ if (recentlyWritten.size() > MAX_WRITTEN_BEFORE) {
+ recentlyWritten.removeFirst();
+ }
+ return obj;
+ }
+ }) {
+ out.writeObject(rdd);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ return new TaskBinary(bytes, instances);
+ }
+
+ /**
+ * The size of the task binary and the classes of the instances it holds.
+ */
+ public static final class TaskBinary {
+ private final long bytes;
+ private final Map<Class<?>, List<String>> instances;
+
+ private TaskBinary(long bytes, Map<Class<?>, List<String>> instances) {
+ this.bytes = bytes;
+ this.instances = instances;
+ }
+
+ public long getBytes() {
+ return bytes;
+ }
+
+ /**
+ * Names of the classes of instances in the task binary that are {@code
className} or extend or
+ * implement it.
+ */
+ public List<String> namesOfSubtypesOf(String className) {
+ return instances.keySet().stream()
+ .filter(clazz -> TaskDeserializationRecorder.isSubtypeOf(clazz,
className))
+ .map(Class::getName)
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * The classes of the objects serialized right before the first instance
of {@code className},
+ * which include the objects that hold it.
+ */
+ public List<String> writtenBefore(String className) {
+ return instances.entrySet().stream()
+ .filter(entry -> entry.getKey().getName().equals(className))
+ .map(Map.Entry::getValue)
+ .findFirst().orElse(Collections.emptyList());
+ }
+ }
+
+ /**
+ * Fails if the task binary holds, or tasks deserialized outside once per
executor deserializations,
+ * an instance of any class that is or extends one of {@code
forbiddenClassNames}, if the scan state
+ * holds an instance of any class that is or extends one of {@code
forbiddenInScanState}, if the task
+ * binary is larger than {@code maxTaskBinaryBytes}, or if a task stream
larger than
+ * {@code maxTaskStreamBytes} was deserialized. Also fails if no Spark RDD
was recorded, which means
+ * the recorder did not see the task binary.
+ */
+ public static void assertTaskDeserializationFootprint(String description,
+
TaskDeserializationRecorder.Result result,
+ TaskBinary taskBinary,
+ Collection<String>
forbiddenClassNames,
+ Collection<String>
forbiddenInScanState,
+ long
maxTaskBinaryBytes,
+ long
maxTaskStreamBytes) {
+ assertEquals(0, result.getErrors(), description + ": the deserialization
recorder failed on some callbacks");
+ assertTrue(!result.namesOfSubtypesOf("org.apache.spark.rdd.RDD").isEmpty(),
+ description + ": no RDD was deserialized on a task thread, so the
recorder did not see the task binary."
+ + " Recorded classes: " + sortedNames(result));
+
assertTrue(!taskBinary.namesOfSubtypesOf("org.apache.spark.rdd.RDD").isEmpty(),
+ description + ": the task binary inspection saw no RDD instance, so it
did not see the task binary");
+ List<String> violations = new ArrayList<>();
+ for (String forbidden : forbiddenClassNames) {
+ List<String> inBinary = taskBinary.namesOfSubtypesOf(forbidden);
+ if (!inBinary.isEmpty()) {
+ violations.add("in the task binary: " + inBinary + " (forbidden: " +
forbidden + "), written right after "
+ + taskBinary.writtenBefore(inBinary.get(0)));
+ }
+ List<String> deserialized = result.namesOfSubtypesOf(forbidden);
+ if (!deserialized.isEmpty()) {
+ violations.add("deserialized per task: " + deserialized + "
(forbidden: " + forbidden + "), first at:\n"
+ + result.getFirstStack(deserialized.get(0)));
+ }
+ }
+ for (String forbidden : forbiddenInScanState) {
+ List<String> inScanState = result.exemptNamesOfSubtypesOf(SCAN_STATE,
forbidden);
+ if (!inScanState.isEmpty()) {
+ violations.add("in the broadcast scan state: " + inScanState + "
(forbidden: " + forbidden + ")");
+ }
+ }
+ if (taskBinary.getBytes() > maxTaskBinaryBytes) {
+ violations.add("the task binary is " + taskBinary.getBytes() + " bytes,
over the budget of " + maxTaskBinaryBytes + " bytes");
+ }
+ if (result.getMaxStreamBytes() > maxTaskStreamBytes) {
+ violations.add("a task stream of " + result.getMaxStreamBytes() + "
bytes was deserialized, over the budget of "
+ + maxTaskStreamBytes + " bytes");
+ }
+ if (!violations.isEmpty()) {
+ fail(description + ": Spark tasks deserialize heavy driver-side state or
more bytes than budgeted with"
+ + " every task; it is paid again by every task. Move it to a
broadcast or out of the closure.\n "
+ + String.join("\n ", violations)
+ + "\nTask binary: " + taskBinary.getBytes() + " bytes (largest task
stream seen by the filter: "
+ + result.getMaxStreamBytes() + " bytes)"
+ + "\nHudi classes deserialized per task: " +
sortedNames(result).stream()
+ .filter(name -> name.startsWith("org.apache.hudi") ||
name.contains(".hudi."))
+ .collect(Collectors.toList()));
+ }
+ }
+
+ private static List<String> sortedNames(TaskDeserializationRecorder.Result
result) {
+ return
result.getClasses().stream().map(Class::getName).sorted().collect(Collectors.toList());
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkReadExecutorFootprint.java
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkReadExecutorFootprint.java
new file mode 100644
index 000000000000..d85101822a9a
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/java/org/apache/hudi/functional/TestSparkReadExecutorFootprint.java
@@ -0,0 +1,597 @@
+/*
+ * 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.hudi.functional;
+
+import org.apache.hudi.DataSourceReadOptions;
+import org.apache.hudi.DataSourceWriteOptions;
+import org.apache.hudi.SparkAdapterSupport$;
+import org.apache.hudi.common.config.HoodieMetadataConfig;
+import org.apache.hudi.common.config.HoodieStorageConfig;
+import org.apache.hudi.common.model.HoodieTableType;
+import org.apache.hudi.common.table.HoodieTableConfig;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.HoodieTableVersion;
+import org.apache.hudi.common.table.timeline.HoodieInstant;
+import org.apache.hudi.common.testutils.HoodieTestTable;
+import org.apache.hudi.common.testutils.InProcessTimeGenerator;
+import org.apache.hudi.config.HoodieArchivalConfig;
+import org.apache.hudi.config.HoodieCompactionConfig;
+import org.apache.hudi.config.HoodieWriteConfig;
+import org.apache.hudi.hadoop.fs.RecordingLocalFileSystem;
+import org.apache.hudi.hadoop.fs.RecordingLocalFileSystem.Call;
+import org.apache.hudi.keygen.SimpleKeyGenerator;
+import org.apache.hudi.testutils.SparkClientFunctionalTestHarness;
+import org.apache.hudi.testutils.SparkExecutorGuards;
+import org.apache.hudi.testutils.TaskDeserializationRecorder;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.spark.SparkConf;
+import org.apache.spark.sql.DataFrameReader;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.SaveMode;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import scala.util.Properties;
+
+import static org.apache.hudi.common.model.HoodieTableType.COPY_ON_WRITE;
+import static org.apache.hudi.common.model.HoodieTableType.MERGE_ON_READ;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Guards what Spark tasks do on the executors when reading a Hudi table
through the file group
+ * reader: they must not access the table's {@code .hoodie} folder, they must
not deserialize heavy
+ * driver-side objects (meta client, timeline, Hadoop configuration, the file
format itself) with
+ * every task, what every task deserializes must stay within a size budget,
and they must not parse
+ * the Hadoop default resources for every file they read. These costs scale
with the number of tasks
+ * or files, not with the data.
+ *
+ * <p>Each table is written once per class and read by every case that needs
it. The table has
+ * several partitions so that a read runs several tasks; on MERGE_ON_READ the
second commit
+ * updates half of the keys so that every file group has log files to merge.
The two commits
+ * follow a history of empty commits, so that every read plans against a
timeline of tens of
+ * instants and per-task state that grows with the timeline shows in the
guards.
+ */
+@Slf4j
+@Tag("functional")
+class TestSparkReadExecutorFootprint extends SparkClientFunctionalTestHarness {
+
+ private static final int CURRENT_VERSION =
HoodieTableVersion.current().versionCode();
+ private static final int NUM_RECORDS = 200;
+ private static final int NUM_UPDATED_RECORDS = 100;
+ private static final int NUM_PARTITIONS = 4;
+ private static final int NUM_HISTORY_COMMITS = 48;
+
+ /**
+ * Driver-side classes that neither a read task nor the broadcast scan state
may deserialize.
+ */
+ private static final List<String> HEAVY_DRIVER_CLASSES = Arrays.asList(
+ "org.apache.hudi.common.table.HoodieTableMetaClient",
+ "org.apache.hudi.common.table.timeline.HoodieTimeline",
+ "org.apache.hudi.storage.StorageConfiguration",
+ "org.apache.spark.util.SerializableConfiguration",
+
"org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReaderBasedFileFormat",
+ "org.apache.hudi.config.HoodieWriteConfig");
+
+ /**
+ * Classes that a read task must not deserialize with its closure or
partition: the heavy driver-side
+ * ones, and the table config and committed instants, which reach the tasks
in the broadcast scan
+ * state that every executor deserializes once. The instants grow with the
timeline.
+ */
+ private static final List<String> CLASSES_NOT_DESERIALIZED_PER_TASK =
Stream.concat(HEAVY_DRIVER_CLASSES.stream(), Stream.of(
+ "org.apache.hudi.common.table.timeline.HoodieInstant",
+ "org.apache.hudi.common.table.read.CommittedInstants",
+
"org.apache.hudi.common.table.HoodieTableConfig")).collect(Collectors.toList());
+
+ /**
+ * Hadoop default resource parses allowed in the tasks of a read. A base
file read converts the table
+ * schema in the scan state with a new configuration once per JVM instance
of the state, so once per
+ * executor, and local mode runs one executor; a merging read does not parse
them.
+ */
+ private static final int MAX_HADOOP_DEFAULT_RESOURCE_LOADS = 1;
+
+ private static final StructType SCHEMA = DataTypes.createStructType(new
StructField[] {
+ DataTypes.createStructField("key", DataTypes.StringType, false),
+ DataTypes.createStructField("part", DataTypes.StringType, false),
+ DataTypes.createStructField("ts", DataTypes.LongType, false),
+ DataTypes.createStructField("value", DataTypes.StringType, true)});
+
+ private static final String SCAN_STATE_CLASS =
"org.apache.spark.sql.execution.datasources.parquet.HoodieFileGroupReadState";
+ private static final String VECTORIZED_READER_ENABLED =
"spark.sql.parquet.enableVectorizedReader";
+ private static final String DATA_SKIPPING_FAILURE_MODE =
"hoodie.fileIndex.dataSkippingFailureMode";
+
+ private static final Map<String, TestTable> TABLES = new HashMap<>();
+
+ @TempDir
+ static Path tablesDir;
+
+ enum TableKind {
+ PLAIN, CDC, PARQUET_LOG_BLOCKS
+ }
+
+ /**
+ * The kind of read a task size budget applies to: a read that returns
columnar batches of base
+ * files measures larger than a read that returns rows.
+ */
+ enum ReadShape {
+ COLUMNAR, ROW
+ }
+
+ /**
+ * What every task of a read may deserialize, by Spark and Scala binary
version: the task binary,
+ * the Java-serialized closure, and the largest task stream, which is the
larger of the binary and
+ * the task with its partition. Any state added to the closure or to a
task's partition is paid
+ * again by every task, whatever the state is. Spark and Scala versions
serialize the plan
+ * differently, so each budget is about 1.1x the largest size measured on
that version, given as
+ * task binary / task stream bytes.
+ */
+ private static final Map<String, TaskBudgets> TASK_BUDGETS = new HashMap<>();
+
+ static {
+ // measured: columnar 15587 / 16679, row 9787 / 12233
+ TASK_BUDGETS.put("3.3_2.12", new TaskBudgets(17152, 18432, 11008, 13568));
+ // measured: columnar 15968 / 17060, row 10040 / 12367
+ TASK_BUDGETS.put("3.4_2.12", new TaskBudgets(17664, 18944, 11264, 13824));
+ // measured: columnar 17555 / 18647, row 11051 / 12488
+ TASK_BUDGETS.put("3.5_2.12", new TaskBudgets(19456, 20736, 12288, 13824));
+ // measured: columnar 17766 / 18989, row 11212 / 12793
+ TASK_BUDGETS.put("3.5_2.13", new TaskBudgets(19712, 20992, 12544, 14080));
+ // measured: columnar 18524 / 19457, row 12434 / 13618
+ TASK_BUDGETS.put("4.0_2.13", new TaskBudgets(20480, 21504, 13824, 15104));
+ // measured: columnar 18491 / 19424, row 12401 / 13618
+ TASK_BUDGETS.put("4.1_2.13", new TaskBudgets(20480, 21504, 13824, 15104));
+ // measured: columnar 18778 / 19712, row 12706 / 13812
+ TASK_BUDGETS.put("4.2_2.13", new TaskBudgets(20736, 21760, 14080, 15360));
+ }
+
+ enum ReadQuery {
+ SNAPSHOT, READ_OPTIMIZED, INCREMENTAL, TIME_TRAVEL, CDC
+ }
+
+ @Override
+ public SparkConf conf() {
+ return conf(Collections.singletonMap("spark.plugins",
SparkExecutorGuards.TASK_START_HOOK_PLUGIN));
+ }
+
+ @BeforeEach
+ void enableRecording() {
+ // Until #20090, a read can leave the session's vectorized reader flag
changed; reset it so that
+ // every case plans its scan the same way whatever ran before it.
+ spark().conf().unset(VECTORIZED_READER_ENABLED);
+
SparkExecutorGuards.enableFileSystemCallRecording(jsc().hadoopConfiguration());
+ }
+
+ @AfterEach
+ void disableRecording() {
+
SparkExecutorGuards.disableFileSystemCallRecording(jsc().hadoopConfiguration());
+ }
+
+ @AfterAll
+ static void forgetTables() {
+ TABLES.clear();
+ }
+
+ static Stream<Arguments> tableVersionsAndTypes() {
+ return Stream.of(6, CURRENT_VERSION).flatMap(version ->
+ Stream.of(COPY_ON_WRITE, MERGE_ON_READ).map(type ->
Arguments.of(version, type)));
+ }
+
+ /**
+ * With the metadata table off, a query other than an incremental one lists
partitions from the
+ * file system with a Spark job, whose tasks are guarded too.
+ */
+ static Stream<Arguments> readQueries() {
+ List<Arguments> args = new ArrayList<>();
+ for (int version : new int[] {6, CURRENT_VERSION}) {
+ for (HoodieTableType type : HoodieTableType.values()) {
+ for (ReadQuery query : Arrays.asList(ReadQuery.SNAPSHOT,
ReadQuery.READ_OPTIMIZED, ReadQuery.INCREMENTAL, ReadQuery.TIME_TRAVEL)) {
+ if (query == ReadQuery.READ_OPTIMIZED && type == COPY_ON_WRITE) {
+ continue;
+ }
+ for (String metadataOnRead : new String[] {"default", "false"}) {
+ args.add(Arguments.of(version, type, query, metadataOnRead));
+ }
+ }
+ }
+ }
+ return args.stream();
+ }
+
+ /**
+ * A CDC read of a version 6 MERGE_ON_READ table fails on the driver, so it
is left out.
+ */
+ static Stream<Arguments> cdcTableVersionsAndTypes() {
+ return tableVersionsAndTypes().filter(args ->
!isVersion6MergeOnRead(args));
+ }
+
+ private static boolean isVersion6MergeOnRead(Arguments args) {
+ Object[] values = args.get();
+ return (int) values[0] == 6 && values[1] == MERGE_ON_READ;
+ }
+
+ static Stream<Arguments> readsForHadoopDefaultResources() {
+ return Stream.of(6, CURRENT_VERSION).flatMap(version -> Stream.of(
+ Arguments.of(version, COPY_ON_WRITE, TableKind.PLAIN),
+ Arguments.of(version, MERGE_ON_READ, TableKind.PLAIN),
+ Arguments.of(version, MERGE_ON_READ, TableKind.PARQUET_LOG_BLOCKS)));
+ }
+
+ /**
+ * Every query on both table versions and types, except the CDC read of a
version 6 MERGE_ON_READ
+ * table, see {@link #cdcTableVersionsAndTypes}.
+ */
+ static Stream<Arguments> readsForDeserialization() {
+ List<Arguments> args = new ArrayList<>();
+ for (int version : new int[] {6, CURRENT_VERSION}) {
+ for (HoodieTableType type : HoodieTableType.values()) {
+ for (ReadQuery query : ReadQuery.values()) {
+ if ((query == ReadQuery.READ_OPTIMIZED && type == COPY_ON_WRITE)
+ || (query == ReadQuery.CDC && version == 6 && type ==
MERGE_ON_READ)) {
+ continue;
+ }
+ TableKind kind = query == ReadQuery.CDC ? TableKind.CDC :
TableKind.PLAIN;
+ args.add(Arguments.of(version, type, kind, query, readShape(type,
query)));
+ }
+ }
+ }
+ return args.stream();
+ }
+
+ /**
+ * A read returns columnar batches when it scans base files without merging:
snapshot and time
+ * travel reads of COPY_ON_WRITE tables, and read optimized reads.
+ */
+ private static ReadShape readShape(HoodieTableType type, ReadQuery query) {
+ switch (query) {
+ case READ_OPTIMIZED:
+ return ReadShape.COLUMNAR;
+ case SNAPSHOT:
+ case TIME_TRAVEL:
+ return type == COPY_ON_WRITE ? ReadShape.COLUMNAR : ReadShape.ROW;
+ default:
+ return ReadShape.ROW;
+ }
+ }
+
+ @ParameterizedTest(name = "[{index}] version={0}, type={1}, query={2},
metadata={3}")
+ @MethodSource("readQueries")
+ void testNoExecutorMetaFolderAccess(int tableVersion, HoodieTableType
tableType, ReadQuery query, String metadataOnRead) {
+ TestTable table = getOrWriteTable(tableVersion, tableType,
TableKind.PLAIN);
+ Map<String, String> options = new HashMap<>();
+ if (!"default".equals(metadataOnRead)) {
+ options.put(HoodieMetadataConfig.ENABLE.key(), metadataOnRead);
+ }
+ List<Row> rows = SparkExecutorGuards.assertNoExecutorMetaFolderAccess(
+ table.name + " " + query + " read with metadata " + metadataOnRead,
+ () -> read(table, query, options).collectAsList());
+ assertFalse(rows.isEmpty(), "The read must return rows for the guard to be
meaningful");
+ if (query == ReadQuery.TIME_TRAVEL) {
+ assertTrue(rows.stream().allMatch(row ->
"v1".equals(row.getAs("value"))),
+ table.name + ": time travel to the first commit must not see the
updates of the second");
+ }
+ }
+
+ @ParameterizedTest(name = "[{index}] version={0}, type={1}")
+ @MethodSource("cdcTableVersionsAndTypes")
+ void testNoExecutorMetaFolderAccessForCdcQuery(int tableVersion,
HoodieTableType tableType) {
+ TestTable table = getOrWriteTable(tableVersion, tableType, TableKind.CDC);
+ List<Row> rows = SparkExecutorGuards.assertNoExecutorMetaFolderAccess(
+ table.name + " CDC read",
+ () -> read(table, ReadQuery.CDC, new HashMap<>()).collectAsList());
+ assertFalse(rows.isEmpty(), "The CDC read must return rows for the guard
to be meaningful");
+ }
+
+ /**
+ * Tasks must not deserialize the meta client, timeline, Hadoop
configuration, write config or the
+ * file format with their closure, and what every task deserializes must
stay within budget.
+ */
+ @ParameterizedTest(name = "[{index}] version={0}, type={1}, table={2},
query={3}, shape={4}")
+ @MethodSource("readsForDeserialization")
+ void testTaskDeserializationFootprint(int tableVersion, HoodieTableType
tableType, TableKind kind, ReadQuery query,
+ ReadShape shape) {
+ String sparkVersion = sparkAndScalaBinaryVersion();
+ TaskBudgets budgets = TASK_BUDGETS.get(sparkVersion);
+ assertNotNull(budgets, "No task size budget for Spark and Scala " +
sparkVersion
+ + "; add one from the task sizes this test logs for each read");
+ TestTable table = getOrWriteTable(tableVersion, tableType, kind);
+ Dataset<Row> df = read(table, query, new HashMap<>());
+ // Plan and list files on the driver first, so that the recorded window
holds only the scan.
+ df.queryExecution().executedPlan().execute();
+ List<Row> rows = new ArrayList<>();
+ TaskDeserializationRecorder.Result result =
SparkExecutorGuards.recordTaskDeserialization(
+ spark().sparkContext(), () -> rows.addAll(df.collectAsList()));
+ assertFalse(rows.isEmpty(), "The read must return rows for the guard to be
meaningful");
+ SparkExecutorGuards.TaskBinary taskBinary =
SparkExecutorGuards.inspectTaskBinary(df);
+ log.info("{} read of {}: task binary {} bytes, largest task stream seen {}
bytes, stages kept {}, ignored {}",
+ query, table.name, taskBinary.getBytes(), result.getMaxStreamBytes(),
result.getKeptScopes(), result.getIgnoredScopes());
+ SparkExecutorGuards.assertTaskDeserializationFootprint(
+ table.name + " " + query + " read on Spark and Scala " + sparkVersion,
result, taskBinary,
+ CLASSES_NOT_DESERIALIZED_PER_TASK, HEAVY_DRIVER_CLASSES,
budgets.maxTaskBinaryBytes(shape), budgets.maxTaskStreamBytes(shape));
+ // Local mode runs one executor, so the broadcast scan state is
deserialized at most once.
+
assertTrue(result.getExemptDeserializations(SparkExecutorGuards.SCAN_STATE,
SCAN_STATE_CLASS) <= 1, table.name + " " + query
+ + " read: the scan state must be deserialized once per executor, not
per task or file, but was deserialized "
+ + result.getExemptDeserializations(SparkExecutorGuards.SCAN_STATE,
SCAN_STATE_CLASS) + " times");
+ }
+
+ /**
+ * Tasks must read base files and log blocks with the configuration they are
given rather than
+ * create one per file, which parses the Hadoop default resources every time.
+ */
+ @ParameterizedTest(name = "[{index}] version={0}, type={1}, table={2}")
+ @MethodSource("readsForHadoopDefaultResources")
+ void testNoHadoopDefaultResourceLoadsPerFile(int tableVersion,
HoodieTableType tableType, TableKind kind) {
+ TestTable table = getOrWriteTable(tableVersion, tableType, kind);
+ Dataset<Row> df = read(table, ReadQuery.SNAPSHOT, new HashMap<>());
+ List<Row> rows =
SparkExecutorGuards.assertTaskHadoopDefaultResourceLoadsAtMost(
+ table.name + " snapshot read", spark().sparkContext(),
MAX_HADOOP_DEFAULT_RESOURCE_LOADS, df::collectAsList);
+ assertEquals(NUM_RECORDS, rows.size());
+ }
+
+ /**
+ * With data skipping, file pruning consults the column stats in the
metadata table on the driver.
+ * The tasks of the scan must still not access {@code .hoodie} of either
table. Every partition holds
+ * the keys {@code i} with the same {@code i % NUM_PARTITIONS}, so a lookup
of the first key prunes the
+ * files of all partitions but the first by their key ranges; strict mode
fails the read if data
+ * skipping cannot be applied. A column stats lookup forced onto the engine
reads the metadata table
+ * from the executors by design and is not covered.
+ */
+ @ParameterizedTest(name = "[{index}] version={0}, type={1}")
+ @MethodSource("tableVersionsAndTypes")
+ void testNoExecutorMetaFolderAccessWithDataSkipping(int tableVersion,
HoodieTableType tableType) {
+ TestTable table = getOrWriteTable(tableVersion, tableType,
TableKind.PLAIN);
+ Map<String, String> options = new HashMap<>();
+ options.put(DataSourceReadOptions.ENABLE_DATA_SKIPPING().key(), "true");
+ List<Row> rows;
+ spark().conf().set(DATA_SKIPPING_FAILURE_MODE, "strict");
+ try {
+ rows = SparkExecutorGuards.assertNoExecutorMetaFolderAccess(
+ table.name + " snapshot read with data skipping",
+ () -> read(table, ReadQuery.SNAPSHOT, options).filter("key = '" +
key(0) + "'").collectAsList());
+ } finally {
+ spark().conf().unset(DATA_SKIPPING_FAILURE_MODE);
+ }
+ assertEquals(1, rows.size());
+
assertTrue(RecordingLocalFileSystem.count(Call.inScope().negate().and(Call.pathContains("/.hoodie/metadata/column_stats/")))
> 0,
+ table.name + ": the driver must read the column stats index for the
read to exercise data skipping");
+
assertTrue(RecordingLocalFileSystem.count(Call.inScope().and(Call.pathContains("/p0/")))
> 0,
+ table.name + ": the tasks must read the files of the partition that
holds the key");
+ Predicate<Call> prunedFileAccess = Call.inScope().and(
+ IntStream.range(1, NUM_PARTITIONS).mapToObj(i ->
Call.pathContains("/p" + i + "/")).reduce(call -> false, Predicate::or));
+ assertEquals(0, RecordingLocalFileSystem.count(prunedFileAccess),
+ table.name + ": data skipping must prune the files of the partitions
without the key, but the tasks read:\n"
+ + RecordingLocalFileSystem.describe(prunedFileAccess));
+ }
+
+ private Dataset<Row> read(TestTable table, ReadQuery query, Map<String,
String> options) {
+ DataFrameReader reader = spark().read().format("hudi").options(options);
+ switch (query) {
+ case SNAPSHOT:
+ reader.option(DataSourceReadOptions.QUERY_TYPE().key(),
DataSourceReadOptions.QUERY_TYPE_SNAPSHOT_OPT_VAL());
+ break;
+ case READ_OPTIMIZED:
+ reader.option(DataSourceReadOptions.QUERY_TYPE().key(),
DataSourceReadOptions.QUERY_TYPE_READ_OPTIMIZED_OPT_VAL());
+ break;
+ case TIME_TRAVEL:
+ reader.option(DataSourceReadOptions.TIME_TRAVEL_AS_OF_INSTANT().key(),
table.firstInstant.requestedTime());
+ break;
+ case INCREMENTAL:
+ case CDC:
+ reader.option(DataSourceReadOptions.QUERY_TYPE().key(),
DataSourceReadOptions.QUERY_TYPE_INCREMENTAL_OPT_VAL())
+ .option(DataSourceReadOptions.START_COMMIT().key(),
table.incrementalTime(table.firstInstant))
+ .option(DataSourceReadOptions.END_COMMIT().key(),
table.incrementalTime(table.lastInstant));
+ if (query == ReadQuery.CDC) {
+ reader.option(DataSourceReadOptions.INCREMENTAL_FORMAT().key(),
DataSourceReadOptions.INCREMENTAL_FORMAT_CDC_VAL());
+ }
+ break;
+ default:
+ throw new IllegalArgumentException("Unknown query " + query);
+ }
+ return reader.load(table.basePath);
+ }
+
+ private TestTable getOrWriteTable(int tableVersion, HoodieTableType
tableType, TableKind kind) {
+ String name = kind.name().toLowerCase() + "_" +
tableType.name().toLowerCase() + "_v" + tableVersion;
+ return TABLES.computeIfAbsent(name, n -> writeTable(n, tableVersion,
tableType, kind));
+ }
+
+ private TestTable writeTable(String name, int tableVersion, HoodieTableType
tableType, TableKind kind) {
+ String basePath = tablesDir.resolve(name).toUri().toString();
+ Map<String, String> options = new HashMap<>();
+ options.put(HoodieTableConfig.NAME.key(), name);
+ options.put(DataSourceWriteOptions.TABLE_TYPE().key(), tableType.name());
+ options.put(DataSourceWriteOptions.RECORDKEY_FIELD().key(), "key");
+ options.put(DataSourceWriteOptions.PARTITIONPATH_FIELD().key(), "part");
+ options.put(DataSourceWriteOptions.ORDERING_FIELDS().key(), "ts");
+ options.put(HoodieWriteConfig.WRITE_TABLE_VERSION.key(),
String.valueOf(tableVersion));
+ options.put(HoodieWriteConfig.AUTO_UPGRADE_VERSION.key(), "false");
+ options.put("hoodie.insert.shuffle.parallelism", "2");
+ options.put("hoodie.upsert.shuffle.parallelism", "2");
+ // Keep the whole history on the active timeline and the log files of
MERGE_ON_READ uncompacted.
+ options.put(HoodieArchivalConfig.MIN_COMMITS_TO_KEEP.key(),
String.valueOf(NUM_HISTORY_COMMITS + 10));
+ options.put(HoodieArchivalConfig.MAX_COMMITS_TO_KEEP.key(),
String.valueOf(NUM_HISTORY_COMMITS + 20));
+ options.put(HoodieCompactionConfig.INLINE_COMPACT_NUM_DELTA_COMMITS.key(),
String.valueOf(NUM_HISTORY_COMMITS + 10));
+ // Index column stats, so that data skipping has an index to consult.
+ options.put(HoodieMetadataConfig.ENABLE_METADATA_INDEX_COLUMN_STATS.key(),
"true");
+ if (kind == TableKind.CDC) {
+ options.put(HoodieTableConfig.CDC_ENABLED.key(), "true");
+ }
+ if (kind == TableKind.PARQUET_LOG_BLOCKS) {
+ options.put(HoodieStorageConfig.LOGFILE_DATA_BLOCK_FORMAT.key(),
"parquet");
+ }
+
+ initTableWithHistory(name, basePath, tableVersion, tableType, kind);
+ List<Row> inserts = IntStream.range(0, NUM_RECORDS)
+ .mapToObj(i -> RowFactory.create(key(i), "p" + (i % NUM_PARTITIONS),
1L, "v1"))
+ .collect(Collectors.toList());
+ write(inserts, SCHEMA, options, basePath);
+ // The second commit updates half of the keys.
+ List<Row> updates = IntStream.range(0, NUM_UPDATED_RECORDS)
+ .mapToObj(i -> RowFactory.create(key(i), "p" + (i % NUM_PARTITIONS),
2L, "v2"))
+ .collect(Collectors.toList());
+ write(updates, SCHEMA, options, basePath);
+
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.builder().setBasePath(basePath).setConf(storageConf()).build();
+ assertEquals(tableVersion,
metaClient.getTableConfig().getTableVersion().versionCode());
+ List<HoodieInstant> commits =
metaClient.getCommitsTimeline().filterCompletedInstants().getInstants();
+ assertEquals(NUM_HISTORY_COMMITS + 2, commits.size(), "Expected the
history and two completed commits in " + name);
+ if (tableType == MERGE_ON_READ) {
+ assertTrue(countLogFiles(tablesDir.resolve(name)) >= NUM_PARTITIONS,
+ "The updates of " + name + " should write log files in every
partition");
+ }
+ return new TestTable(name, basePath, tableVersion,
commits.get(NUM_HISTORY_COMMITS), commits.get(NUM_HISTORY_COMMITS + 1));
+ }
+
+ /**
+ * Creates the table with the configuration the writes use and {@link
#NUM_HISTORY_COMMITS} empty
+ * completed commits.
+ */
+ private void initTableWithHistory(String name, String basePath, int
tableVersion, HoodieTableType tableType,
+ TableKind kind) {
+ try {
+ HoodieTableMetaClient metaClient =
HoodieTableMetaClient.newTableBuilder()
+ .setTableType(tableType)
+ .setTableName(name)
+ .setTableVersion(tableVersion)
+ .setRecordKeyFields("key")
+ .setPartitionFields("part")
+ .setOrderingFields("ts")
+ .setKeyGeneratorClassProp(SimpleKeyGenerator.class.getName())
+ .setCDCEnabled(kind == TableKind.CDC)
+ .initTable(storageConf(), basePath);
+ HoodieTestTable history = HoodieTestTable.of(metaClient);
+ for (int i = 0; i < NUM_HISTORY_COMMITS; i++) {
+ String instantTime = InProcessTimeGenerator.createNewInstantTime();
+ if (tableType == COPY_ON_WRITE) {
+ history.addCommit(instantTime);
+ } else {
+ history.addDeltaCommit(instantTime);
+ }
+ }
+ } catch (Exception e) {
+ throw new IllegalStateException("Cannot create the history of " + name,
e);
+ }
+ }
+
+ private void write(List<Row> rows, StructType schema, Map<String, String>
options, String basePath) {
+ spark().createDataset(rows,
SparkAdapterSupport$.MODULE$.sparkAdapter().getCatalystExpressionUtils().getEncoder(schema))
+ .write()
+ .format("hudi")
+ .options(options)
+ .mode(SaveMode.Append)
+ .save(basePath);
+ }
+
+ private static long countLogFiles(Path tableDir) {
+ try (Stream<Path> files = Files.walk(tableDir)) {
+ return files.filter(p -> !p.toString().contains("/.hoodie/") &&
p.getFileName().toString().contains(".log.")).count();
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ /**
+ * The running Spark version and Scala binary version, such as {@code
3.5_2.12}.
+ */
+ private String sparkAndScalaBinaryVersion() {
+ String[] spark = spark().version().split("\\.");
+ String[] scala = Properties.versionNumberString().split("\\.");
+ return spark[0] + "." + spark[1] + "_" + scala[0] + "." + scala[1];
+ }
+
+ private static String key(int i) {
+ return String.format("key%03d", i);
+ }
+
+ private static final class TaskBudgets {
+ private final long columnarTaskBinaryBytes;
+ private final long columnarTaskStreamBytes;
+ private final long rowTaskBinaryBytes;
+ private final long rowTaskStreamBytes;
+
+ private TaskBudgets(long columnarTaskBinaryBytes, long
columnarTaskStreamBytes, long rowTaskBinaryBytes,
+ long rowTaskStreamBytes) {
+ this.columnarTaskBinaryBytes = columnarTaskBinaryBytes;
+ this.columnarTaskStreamBytes = columnarTaskStreamBytes;
+ this.rowTaskBinaryBytes = rowTaskBinaryBytes;
+ this.rowTaskStreamBytes = rowTaskStreamBytes;
+ }
+
+ private long maxTaskBinaryBytes(ReadShape shape) {
+ return shape == ReadShape.COLUMNAR ? columnarTaskBinaryBytes :
rowTaskBinaryBytes;
+ }
+
+ private long maxTaskStreamBytes(ReadShape shape) {
+ return shape == ReadShape.COLUMNAR ? columnarTaskStreamBytes :
rowTaskStreamBytes;
+ }
+ }
+
+ private static final class TestTable {
+ private final String name;
+ private final String basePath;
+ private final int tableVersion;
+ private final HoodieInstant firstInstant;
+ private final HoodieInstant lastInstant;
+
+ private TestTable(String name, String basePath, int tableVersion,
HoodieInstant firstInstant, HoodieInstant lastInstant) {
+ this.name = name;
+ this.basePath = basePath;
+ this.tableVersion = tableVersion;
+ this.firstInstant = firstInstant;
+ this.lastInstant = lastInstant;
+ }
+
+ /**
+ * Incremental queries range over requested times before table version 8
and over completion
+ * times from version 8 on.
+ */
+ private String incrementalTime(HoodieInstant instant) {
+ return tableVersion < 8 ? instant.requestedTime() :
instant.getCompletionTime();
+ }
+ }
+}
