HyukjinKwon commented on code in PR #59932:
URL: https://github.com/apache/spark/pull/59932#discussion_r4417634702
##########
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:
Done in 8b94b3b: the batch and the stream writers record a successful
commit, and `abort` skips the library after it. The order of the calls in the
header now says that Spark does not abort a writer or a data writer whose
commit succeeded.
##########
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:
Done in 8b94b3b: `NativeDataWriter` records a failed write or commit, after
which `write` and `commit` fail with an internal error, and only `abort` and
`close` reach the library.
##########
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:
Done in 8b94b3b: the test checks TRACE, DEBUG, INFO, WARN and ERROR.
##########
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:
Done in 8b94b3b: `ColumnarScanBuilder.pruneColumns` asks the planner at most
once. A later call with columns outside of the read schema fails with an
internal error.
##########
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:
Done in 8b94b3b: the test library records the calls that the convention
rules out (on writers, data writers and batch readers) in a file,
`NativeDataSourceTestUtils.orderViolations`, that `NativeDataSourceSuite`
checks after each test. Letting a data writer be aborted after a successful
commit makes "do not abort a data writer that committed with an invalid
message" fail.
--
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]
