dongjoon-hyun commented on code in PR #59932:
URL: https://github.com/apache/spark/pull/59932#discussion_r4417356082
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -880,18 +925,14 @@ class NativeDataWriter(
val root = new VectorSchemaRoot(
vectors.map(_.getField).asJava, vectors.asJava, batch.numRows())
val array = ArrowArray.allocateNew(ArrowUtils.rootAllocator)
- val arrowSchema = ArrowSchema.allocateNew(ArrowUtils.rootAllocator)
try {
- Data.exportVectorSchemaRoot(ArrowUtils.rootAllocator, root, null, array,
arrowSchema)
- call(name, "write to") {
- library.write(handle, array.memoryAddress(),
arrowSchema.memoryAddress())
- }
+ // Only the array: the data writer got the schema when it was created.
+ Data.exportVectorSchemaRoot(ArrowUtils.rootAllocator, root, null, array)
+ call(name, "write to")(library.write(handle, array.memoryAddress()))
Review Comment:
**2. [Medium] The adapter does not enforce the order after a failed call**
With this PR, "after a function on a handle fails, Spark calls no other
function on the handle than its close function" becomes part of version 1,
whose semantics never change, so libraries will drop their own guards. For the
data writer, it holds only because `DataWritingSparkTask` calls it in the right
order. `finished` stays false when `commit_data_writer` fails, so that `abort`
still runs, but then the `finished || closed` checks of `write` and `commit`
also let another `write` or `commit` reach the library. If a caller ever
retries a commit, or writes after a failure, the library gets a call that the
header rules out, and Spark reports nothing.
Could the data writer record that a call failed, for example with a flag set
when `write` or `commitDataWriter` throws, so that `write` and `commit` throw
an internal error after it and only `abort` and `close` go through? The same
idea applies to the other handles: see item 1 for the stream writer and item 4
for `prune_columns`.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -426,7 +447,8 @@ class NativeDataSourceStreamWriter(
partitionId: Int,
taskId: Long,
epochId: Long): DataWriter[ColumnarBatch] = {
- new NativeDataWriter(location, name, writerState, partitionId, taskId,
epochId)
+ new NativeDataWriter(
+ location, name, schema, timeZoneId, writerState, partitionId, taskId,
epochId)
}
override def commit(epochId: Long, messages: Array[WriterCommitMessage]):
Unit = {
Review Comment:
**1. [Medium] A stream writer can be aborted after its commit succeeded**
The new convention says that Spark aborts a writer whose commit failed, but
not whether it can abort one whose commit succeeded, and here it can.
`NativeDataSourceWriter.commit` closes the handle once the library commits, so
a later `abort` fails on `handle.get` before it reaches the library. This
`commit` neither closes the handle nor records that it succeeded.
In `V2TableWriteExec.writeWithV2`, the code after
`batchWrite.commit(messages)` returns is still inside the `try` (a `logInfo`
and `commitProgress = Some(...)`), and `case cause: Throwable` then calls
`batchWrite.abort(messages)`, which `MicroBatchWrite` forwards to the `abort`
below, for the epoch that the library just committed. Only a fatal error, such
as an `OutOfMemoryError`, gets there today, but a sink that removes the output
of an epoch on abort would then lose committed data.
Could `commit` record its success, and `abort` skip the library after it, as
`NativeDataWriter` does with `finished`? The header could then say for writers
what it says for `commit_data_writer`: if commit succeeds, Spark does not call
abort.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -1114,6 +1185,19 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
message.contains("a message from native code") &&
message.contains(System.mapLibraryName("test_native_datasource"))
})
+
+ // is_log_enabled follows the level of the log of Spark: the library gets
whether Spark logs
+ // each level, from error to trace.
+ val levels = new File(Utils.createTempDir(), "levels")
+ Seq(
+ org.apache.logging.log4j.Level.DEBUG -> "1,1,1,1,0",
+ org.apache.logging.log4j.Level.ERROR -> "1,0,0,0,0").foreach { case
(level, expected) =>
Review Comment:
**3. [Medium] The test cannot tell WARN, INFO and DEBUG apart**
At DEBUG every level from error to debug is enabled, and at ERROR every
level from warn on is disabled, so these two cases only check error against the
rest and debug against trace. If `isLogEnabled` mapped `2` to
`isInfoEnabled()`, or `3` to `isWarnEnabled()`, both cases would still pass,
although libraries would get wrong answers at INFO, the default level. Checking
the other levels too would cover each case of the switch:
```suggestion
org.apache.logging.log4j.Level.TRACE -> "1,1,1,1,1",
org.apache.logging.log4j.Level.DEBUG -> "1,1,1,1,0",
org.apache.logging.log4j.Level.INFO -> "1,1,1,0,0",
org.apache.logging.log4j.Level.WARN -> "1,1,0,0,0",
org.apache.logging.log4j.Level.ERROR -> "1,0,0,0,0").foreach { case
(level, expected) =>
```
`NativeLibrary` has no logger configuration of its own, so `withLogAppender`
sets these levels on the root logger configuration: TRACE applies to all of
Spark while `range` runs. It can be left out if it makes the test slow.
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -115,8 +117,14 @@
* released once they are no longer referenced (see Lifecycle above). A
reader or writer handle
* must not depend on the data source handle it was created from, because
Spark closes the data
* source handle right after creating it.
- * - Memory. What Spark passes to a function, such as strings, byte arrays,
SparkDsCreateInfo and
- * SparkDsWriteInfo, is only valid during the call, except for SparkDsHost
and
+ * - Order of the calls. After a function on a handle fails, Spark calls no
other function on the
+ * handle than its close function, except that it aborts a writer whose
commit failed and a data
+ * writer whose write or commit failed. It calls each function of a batch
reader at most once. It
+ * calls commit and abort at most once each for a writer, and
commit_data_writer and
+ * abort_data_writer at most once each for a data writer, with all the calls
to write before
+ * them.
Review Comment:
**5. [Low] No test checks the order of the calls**
Since the convention becomes part of version 1, a test would keep a later
change in Spark from breaking it silently, for example a retry in
`DataWritingSparkTask`, or a stream writer aborted after its commit (item 1).
The test library could keep a small state for each handle (failed, committed,
aborted, planned) and report any call that the convention rules out, such as
"abort_data_writer called after commit_data_writer succeeded", for instance by
recording it in a file that the suite checks. The existing tests that inject
failures with the option `fail` would then check the convention without new
test cases.
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -284,6 +294,13 @@ typedef struct SparkDsHost {
* of the driver or of the executor, with the path of the library.
*/
void (*log)(int32_t level, const char* message);
+ /*
+ * Returns whether log writes the messages of one of the SPARK_DS_LOG_*
levels, so that the
+ * library only formats the messages that Spark writes, and gets those that
Spark writes at the
+ * levels that a library would not log by default, such as debug. It follows
the configuration of
+ * the log of Spark, which can change while Spark runs.
+ */
+ bool (*is_log_enabled)(int32_t level);
Review Comment:
**7. [Low] Levels outside `SPARK_DS_LOG_*` are left unspecified**
`isLogEnabled` answers for any other level as for `SPARK_DS_LOG_INFO`, like
`log`, which logs such messages as info, but neither comment says so. Since
version 1 is frozen once 4.4.0 ships, it may help to state it for both
functions (for example "other levels are treated as SPARK_DS_LOG_INFO"), so
that SDKs know what a level of 0 or 6 does and a later release cannot change it.
Two smaller points about the same comment: it could say that the pointer is
never NULL in a struct large enough to hold it, and "whether log writes the
messages" is a little stronger than what Spark checks, which is the level of
its logger: a filter on an appender can still drop a message for which this
returns true.
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -115,8 +117,14 @@
* released once they are no longer referenced (see Lifecycle above). A
reader or writer handle
* must not depend on the data source handle it was created from, because
Spark closes the data
* source handle right after creating it.
- * - Memory. What Spark passes to a function, such as strings, byte arrays,
SparkDsCreateInfo and
- * SparkDsWriteInfo, is only valid during the call, except for SparkDsHost
and
+ * - Order of the calls. After a function on a handle fails, Spark calls no
other function on the
+ * handle than its close function, except that it aborts a writer whose
commit failed and a data
+ * writer whose write or commit failed. It calls each function of a batch
reader at most once. It
Review Comment:
**4. [Low] `prune_columns` is called at most once only because Spark calls
`pruneColumns` once**
"It calls each function of a batch reader at most once" holds today:
`ColumnarScanBuilder.pushLimit` asks the reader only once (`limitResult`),
`ColumnarBatchScan` plans the partitions once, and `NativeHandle` releases the
reader once. `ColumnarScanBuilder.pruneColumns`, though, forwards every call to
`batchPlanner.pruneColumns`, so the guarantee rests on `V2ScanRelationPushDown`
pruning each scan builder once. It does so now, but the Data Source V2 API does
not promise it. A once-flag like `limitResult` would keep the convention true
whatever the optimizer does.
##########
docs/sql-native-data-sources.md:
##########
@@ -70,7 +70,9 @@ A native data source library exports a single C function,
`spark_ds_init`, which
of function pointers, a `SparkDsLibrary`, that implements the data source: it
needs no JVM headers
or bindings. Each library has its own table, so any number of native data
sources can be loaded
side by side. When Spark loads a library, it calls `spark_ds_init` with a
`SparkDsHost`, which
-gives the library the version of Spark and a function to log messages to the
log of Spark.
+gives the library the version of Spark, a function to log messages to the log
of Spark, and a
+function that tells whether Spark logs the messages of a level, so that a
library follows the log
+level of Spark.
Review Comment:
**8. [Nit] The guide does not mention the order of the calls**
This paragraph gets the new function, but the convention, which the
description presents as what lets SDKs drop their guards, is only in the
header. The guide already states the other conventions further down ("Each
handle that a `create` function returns is released exactly once, by the
`close` function of its kind"), so a sentence there would also reach the
readers of the Rust and C++ examples.
##########
sql/core/src/main/java/org/apache/spark/sql/execution/datasources/v2/ffi/NativeLibrary.java:
##########
@@ -491,6 +546,25 @@ private static void log(String path, int level, long
message) {
}
}
+ /**
+ * Called by the library, through {@code SparkDsHost.is_log_enabled}, with
the level of the
+ * messages of {@link #log}, which logs the levels that it does not know as
info. It must not
+ * throw: an exception thrown by an upcall crashes the JVM.
+ */
+ private static boolean isLogEnabled(int level) {
+ try {
+ return switch (level) {
Review Comment:
**9. [Nit] `isLogEnabled` repeats the level mapping of `log`**
Both switches map 1, 2, 4 and 5 and treat the rest as info, so they have to
change together, and with the current test (item 3) a mismatch would go
unnoticed. Could they share the mapping, for example through a small helper
that turns the level into an enum, or at least point at each other in a comment?
##########
sql/core/src/test/resources/native-datasource/test_native_datasource.cc:
##########
@@ -648,14 +656,34 @@ thread_local std::string last_error;
// Spark, as spark_ds_init got it.
const SparkDsHost* host = nullptr;
-// Logs with the log of Spark. A library must only use the fields that fit in
the size of the
-// struct that Spark passes, which may be of another version of the header.
+// Whether the struct that Spark passes has is_log_enabled, which an earlier
version of the header
+// does not have.
+bool hasIsLogEnabled() {
+ return host->size >= static_cast<int64_t>(offsetof(SparkDsHost,
is_log_enabled) +
+ sizeof(host->is_log_enabled));
+}
+
+// Logs with the log of Spark, unless it does not write the level. A library
must only use the
+// fields that fit in the size of the struct that Spark passes, which may be
of another version of
+// the header.
void hostLog(int32_t level, const std::string& message) {
+ if (hasIsLogEnabled() && !host->is_log_enabled(level)) return;
Review Comment:
**6. [Low] No test checks that `hostLog` skips the levels that Spark does
not log**
The description says that the test library now skips those levels, but the
only check of `hostLog` logs at WARN with the logger at INFO, where WARN is
enabled, so removing this line would not fail any test. Logging with the option
`log` while the logger is at ERROR, and checking that the appender gets no WARN
event, would cover it.
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -284,6 +294,13 @@ typedef struct SparkDsHost {
* of the driver or of the executor, with the path of the library.
*/
void (*log)(int32_t level, const char* message);
+ /*
+ * Returns whether log writes the messages of one of the SPARK_DS_LOG_*
levels, so that the
+ * library only formats the messages that Spark writes, and gets those that
Spark writes at the
+ * levels that a library would not log by default, such as debug. It follows
the configuration of
+ * the log of Spark, which can change while Spark runs.
Review Comment:
**10. [Nit] Each check is an upcall, and the comment discourages caching it**
"which can change while Spark runs" tells a library to ask before every
message, but each call goes from native code into the JVM. An SDK whose
`log::Log::enabled` (in Rust) or logging macro calls it before every `debug!`
in a loop over batches pays that transition even when debug is off, which is
not free either. It may help to say that a library may cache the answer and
refresh it from time to time, so that SDKs can keep hot paths free of upcalls.
--
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]
