HyukjinKwon commented on code in PR #59932:
URL: https://github.com/apache/spark/pull/59932#discussion_r4418128358
##########
sql/core/src/main/resources/org/apache/spark/sql/datasource/spark_datasource.h:
##########
@@ -115,8 +118,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 does not abort a writer or a data
writer whose commit
+ * succeeded. It calls each function of a batch reader at most once. It
calls commit and abort at
Review Comment:
Done in 0d5c9c2: the Writes paragraph now says that Spark aborts the data
writers of the other attempts except two, one whose `commit_data_writer`
succeeded before its task failed, which is not aborted, and one on an executor
that is lost, and that neither is passed to `commit`, so the commit or the
abort of the writer cleans up the output of the data writers whose messages it
does not get. The sentence before it says that `abort_data_writer` follows a
failure only before `commit_data_writer` succeeded.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -1114,6 +1236,30 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
message.contains("a message from native code") &&
message.contains(System.mapLibraryName("test_native_datasource"))
})
+ // The library does not log a warning when Spark logs errors only.
+ val errorAppender = new LogAppender("native data source error logs")
+ withLogAppender(errorAppender, loggerNames =
Seq(classOf[NativeLibrary].getName),
+ level = Some(org.apache.logging.log4j.Level.ERROR)) {
+ range(session, "log" -> "a message that Spark does not log")
+ }
+ assert(!errorAppender.loggingEvents.exists(
+ _.getMessage.getFormattedMessage.contains("a message that Spark does not
log")))
Review Comment:
Done in 0d5c9c2: with the option `log_calls_path`, `hostLog` appends the
level to that file right before it calls `host->log`. The test checks that the
file has the warning at the default level, and that it does not exist at ERROR.
Removing the check of `is_log_enabled` in `hostLog` now fails the test.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSource.scala:
##########
@@ -376,35 +385,49 @@ object NativeDataSourceStreamReader {
JsonMapper.builder().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS).build()
}
-/** Writes to a native data source for a batch write. */
+/**
+ * Writes to a native data source for a batch write.
+ *
+ * @param schema the schema of the data to write
+ * @param timeZoneId the time zone of the timestamps of the schema that the
writer got
+ */
class NativeDataSourceWriter(
location: NativeLibraryLocation,
name: String,
+ schema: StructType,
+ timeZoneId: String,
@transient private val handle: NativeHandle,
writerState: Array[Byte])
extends ColumnarWriter {
import NativeCalls._
@transient private lazy val library = NativeLibraries.get(location)
+ // Whether the library committed the write, after which it is not aborted,
as the header says,
+ // even if Spark fails afterwards.
+ @transient private var committed = false
Review Comment:
Done in 0d5c9c2: "keep the order of the calls whatever the callers of the
adapter do" aborts a committed batch write and a committed micro-batch, writes
and commits a data writer after its write failed, and prunes the columns of a
scan builder twice and then with another column. Removing the committed check
of `ColumnarBatchWrite`, the `failed` check of `writeOrCommit`, or the `pruned`
guard each fails it.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceSuite.scala:
##########
@@ -100,6 +100,19 @@ class NativeDataSourceSuite extends QueryTest with
SharedSparkSession {
session
}
+ // The test libraries record the calls that the order of the calls of the
header rules out.
+ override def afterEach(): Unit = {
+ try {
+ if (orderViolations.exists()) {
+ val violations = Files.readString(orderViolations.toPath)
+ orderViolations.delete()
+ fail(s"The test library got calls that the header rules
out:\n$violations")
Review Comment:
Done in 0d5c9c2: `checkingOrderOfCalls` in `NativeDataSourceTestUtils`
deletes the file before the test and fails the test itself when the file exists
after it. `nativeTest` uses it, and `afterEach` is gone.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/ffi/NativeDataSourceTestUtils.scala:
##########
@@ -37,6 +37,12 @@ object NativeDataSourceTestUtils {
private val libraries = new ConcurrentHashMap[String, File]()
+ /**
+ * The file to which the test libraries append the calls that the order of
the calls of the
+ * header rules out, one per line. It does not exist as long as there are
none.
+ */
+ lazy val orderViolations: File = new File(buildDir, "order_violations")
Review Comment:
Done in 0d5c9c2: `NativeDataSourceClusterSuite` and
`NativeDataSourceConnectSuite` run their tests with `checkingOrderOfCalls` too.
--
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]
