feat: support spark jdbc write to duckdb.

[chaozwn]

spark uses path to store catalog information, which is the option injected by spark by default. duckdb-java is recommended to add the path parameter recognition, at least do not throw RuntimeException.

16:28:57    Runtime Error in model movies (models/sources/movies.sql)
  An error occurred while calling o40.sql.
  : java.sql.SQLException: Invalid Input Error: Unrecognized configuration property "path"
        at org.duckdb.DuckDBNative.duckdb_jdbc_startup(Native Method)
        at org.duckdb.DuckDBConnection.newConnection(DuckDBConnection.java:48)
        at org.duckdb.DuckDBDriver.connect(DuckDBDriver.java:41)
        at org.apache.spark.sql.execution.datasources.jdbc.connection.BasicConnectionProvider.getConnection(BasicConnectionProvider.scala:49)
        at org.apache.spark.sql.execution.datasources.jdbc.connection.ConnectionProviderBase.create(ConnectionProvider.scala:102)
        at org.apache.spark.sql.jdbc.JdbcDialect.$anonfun$createConnectionFactory$1(JdbcDialects.scala:123)
        at org.apache.spark.sql.jdbc.JdbcDialect.$anonfun$createConnectionFactory$1$adapted(JdbcDialects.scala:119)
        at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:50)
        at org.apache.spark.sql.execution.datasources.DataSource.writeAndRead(DataSource.scala:506)
        at org.apache.spark.sql.execution.command.CreateDataSourceTableAsSelectCommand.saveDataIntoTable(createDataSourceTables.scala:228)
        at org.apache.spark.sql.execution.command.CreateDataSourceTableAsSelectCommand.run(createDataSourceTables.scala:183)
        at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:75)
        at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:73)
        at org.apache.spark.sql.execution.command.ExecutedCommandExec.executeCollect(commands.scala:84)
        at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.$anonfun$applyOrElse$1(QueryExecution.scala:98)
        at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$6(SQLExecution.scala:118)
        at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:195)
        at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:103)
        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:827)
        at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:65)
        at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:98)
        at org.apache.spark.sql.execution.QueryExecution$$anonfun$eagerlyExecuteCommands$1.applyOrElse(QueryExecution.scala:94)
        at org.apache.spark.sql.catalyst.trees.TreeNode.$anonfun$transformDownWithPruning$1(TreeNode.scala:512)
        at org.apache.spark.sql.catalyst.trees.CurrentOrigin$.withOrigin(TreeNode.scala:104)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDownWithPruning(TreeNode.scala:512)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.org$apache$spark$sql$catalyst$plans$logical$AnalysisHelper$$super$transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning(AnalysisHelper.scala:267)
        at org.apache.spark.sql.catalyst.plans.logical.AnalysisHelper.transformDownWithPruning$(AnalysisHelper.scala:263)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.transformDownWithPruning(LogicalPlan.scala:31)
        at org.apache.spark.sql.catalyst.trees.TreeNode.transformDown(TreeNode.scala:488)
        at org.apache.spark.sql.execution.QueryExecution.eagerlyExecuteCommands(QueryExecution.scala:94)
        at org.apache.spark.sql.execution.QueryExecution.commandExecuted$lzycompute(QueryExecution.scala:81)
        at org.apache.spark.sql.execution.QueryExecution.commandExecuted(QueryExecution.scala:79)
        at org.apache.spark.sql.Dataset.<init>(Dataset.scala:219)
        at org.apache.spark.sql.Dataset$.$anonfun$ofRows$2(Dataset.scala:99)
        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:827)
        at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:96)
        at org.apache.spark.sql.SparkSession.$anonfun$sql$1(SparkSession.scala:640)
        at org.apache.spark.sql.SparkSession.withActive(SparkSession.scala:827)
        at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:630)
        at org.apache.spark.sql.SparkSession.sql(SparkSession.scala:662)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:566)
        at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
        at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374)
        at py4j.Gateway.invoke(Gateway.java:282)
        at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
        at py4j.commands.CallCommand.execute(CallCommand.java:79)
        at py4j.ClientServerConnection.waitForCommands(ClientServerConnection.java:182)
        at py4j.ClientServerConnection.run(ClientServerConnection.java:106)
        at java.base/java.lang.Thread.run(Thread.java:829)

[lanklaas]

I got the same issue and got around it by patching the driver like this in the DuckDBDriver.java connect method

Add a list of properties as static in the top:

    public static final String[] SPARK_PROPS = {"path","dbtable","serialization.format","fetchSize"};

Then in the connect method before the read only property is removed:

for (String prop : SPARK_PROPS) {
    info.remove(prop);
}

ran make release and now it works with my spark (3.5.2)

[staticlibs]

@chaozwn

Hi! I would like to fix this, just can you clarify how exactly you are running Spark to make it pass the path connection option?

For example, the following with Spark 3.5.5 SQL shell works correctly for me:

$ ./bin/spark-sql --driver-class-path path/to/duckdb_jdbc.jar --driver-class-path path/to/duckdb_jdbc.jar
> spark-sql (default)> CREATE TEMPORARY VIEW jdbcTable
                   > USING org.apache.spark.sql.jdbc
                   > OPTIONS (
                   >     url "jdbc:duckdb:",
                   >     query "select * from 'path/to/taxi_2019_04.parquet'"
                   > );
Time taken: 1.883 seconds
spark-sql (default)> select count(*) from jdbcTable;
7433139
Time taken: 2.382 seconds, Fetched 1 row(s)

[lanklaas]

@staticlibs to get the path error you can create a table:

Started spark like this:

/t/spark-3.5.5-bin-hadoop3 (local)> ./bin/spark-sql --driver-class-path /tmp/duckdb_jdbc.jar 

CREATE TABLE test
USING org.apache.spark.sql.jdbc OPTIONS (
    url "jdbc:duckdb:/tmp/test.ddb",
    dbtable "main.test_data",
    driver "org.duckdb.DuckDBDriver",
    fetchSize "1000"
);

This throws:

java.sql.SQLException: Invalid Input Error: The following options were not recognized: path

[staticlibs]

@lanklaas

Thanks for the details! Just perhaps I am missing something about my Spark setup, but I cannot reproduce the problem:

alex@localhost:~/projects/duck/spark/spark-3.5.5-bin-hadoop3$ rm -rf ./metastore_db/
alex@localhost:~/projects/duck/spark/spark-3.5.5-bin-hadoop3$ ./bin/spark-sql --driver-class-path ~/.m2/repository/org/duckdb/duckdb_jdbc/1.0.0/duckdb_jdbc-1.0.0.jar --conf spark.driver.extraJavaOptions="-agentlib:jdwp=transport=dt_socket,server=y,suspend=n,address=5005"
Listening for transport dt_socket at address: 5005
25/04/03 19:27:16 WARN Utils: Your hostname, localhost.localdomain resolves to a loopback address: 127.0.0.1; using 192.168.178.86 instead (on interface wlp61s0)
25/04/03 19:27:16 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
25/04/03 19:27:17 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
25/04/03 19:27:20 WARN HiveConf: HiveConf of name hive.stats.jdbc.timeout does not exist
25/04/03 19:27:20 WARN HiveConf: HiveConf of name hive.stats.retries.wait does not exist
25/04/03 19:27:25 WARN ObjectStore: Version information not found in metastore. hive.metastore.schema.verification is not enabled so recording the schema version 2.3.0
25/04/03 19:27:25 WARN ObjectStore: setMetaStoreSchemaVersion called but recording version is disabled: version = 2.3.0, comment = Set by MetaStore alex@127.0.0.1
25/04/03 19:27:25 WARN ObjectStore: Failed to get database default, returning NoSuchObjectException
Spark Web UI available at http://192.168.178.86:4040
Spark master: local[*], Application Id: local-1743704838978
spark-sql (default)> CREATE TABLE test
                   > USING org.apache.spark.sql.jdbc OPTIONS (
                   >     url "jdbc:duckdb:test.db",
                   >     dbtable "test_data",
                   >     driver "org.duckdb.DuckDBDriver",
                   >     fetchSize "1000"
                   > );
25/04/03 19:28:28 WARN HiveExternalCatalog: Couldn't find corresponding Hive SerDe for data source provider org.apache.spark.sql.jdbc. Persisting data source table `spark_catalog`.`default`.`test` into Hive metastore in Spark SQL specific format, which is NOT compatible with Hive.
25/04/03 19:28:28 WARN SessionState: METASTORE_FILTER_HOOK will be ignored, since hive.security.authorization.manager is set to instance of HiveAuthorizerFactory.
25/04/03 19:28:28 WARN HiveConf: HiveConf of name hive.internal.ss.authz.settings.applied.marker does not exist
25/04/03 19:28:28 WARN HiveConf: HiveConf of name hive.stats.jdbc.timeout does not exist
25/04/03 19:28:28 WARN HiveConf: HiveConf of name hive.stats.retries.wait does not exist
Time taken: 54.79 seconds

I can see in debugger that no properties are passed at all to DuckDBDriver#connect() method.

[lanklaas]

@staticlibs Sorry, I forgot to mention that the issue happens when you select from the table. In case there are some compatibility issues I made this dockerfile that should get rid of them:

# This version of debian gets an older version of libc (2.31)
# that is more compatible with older and newer code
FROM openjdk:17-jdk-bullseye as builder

RUN apt update
RUN apt install -y build-essential
RUN apt install -y cmake

WORKDIR /source
RUN wget https://github.com/duckdb/duckdb-java/archive/refs/tags/v1.2.1.tar.gz

RUN tar -xzf v1.2.1.tar.gz

WORKDIR /source/duckdb-java-1.2.1

RUN make release

FROM openjdk:17-jdk-bullseye

WORKDIR /opt

RUN wget https://dlcdn.apache.org/spark/spark-3.5.5/spark-3.5.5-bin-hadoop3.tgz

RUN tar -xvzf spark-3.5.5-bin-hadoop3.tgz

RUN rm spark-3.5.5-bin-hadoop3.tgz

RUN curl https://install.duckdb.org | sh

RUN echo "CREATE TABLE test_data AS SELECT range AS id, 'user_' || range::VARCHAR AS username, 20 + (range % 30) AS age, cast(NOW() as timestamp) - INTERVAL (range || ' days') AS created_at FROM range(500);" | /root/.duckdb/cli/latest/duckdb /tmp/test.ddb

COPY --from=builder /source/duckdb-java-1.2.1/build/release/duckdb_jdbc.jar /tmp/

docker build -t test .
docker run -it test bash
cd spark-3.5.5-bin-hadoop3/

CREATE TABLE test
USING org.apache.spark.sql.jdbc OPTIONS (
    url "jdbc:duckdb:/tmp/test.ddb",
    dbtable "main.test_data",
    driver "org.duckdb.DuckDBDriver",
    fetchSize "1000"
);
select * from test;

[staticlibs]

@lanklaas

Thanks! Select reproduces it. That the path option is passed seems to be a bug in Spark (at least according to this comment). But there seems to be no easy way to report it to Spark (no GH bugtracker).

I've added ignoring the path in #184. I see that you also proposed to ignore other possible Spark options: dbtable, serialization.format, fetchSize. Quick search on Spark sources does not show that they can be passed, so I am not adding them to ignore list now. If there are scenarios when they are actually passed - I will be able to fix that promptly.