This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 4c6bdcf8de Issue #8138 : Do not send a path option when writing with
file_format=jdbc (#8176)
4c6bdcf8de is described below
commit 4c6bdcf8de670ec92ef73636d113ee029126081a
Author: vbhanuchander-lang <[email protected]>
AuthorDate: Mon Aug 31 03:40:48 2026 -0400
Issue #8138 : Do not send a path option when writing with file_format=jdbc
(#8176)
SparkFileIoSupport.writeDataset always called DataFrameWriter.save(path).
That one-argument overload stores its argument as the "path" writer option
before saving, and Spark's JDBC provider forwards every option it does not
recognise to the driver as a connection property. A JDBC sink has no path,
so the driver receives one it never asked for.
Postgres and MySQL ignore properties they do not know, so nothing surfaces
there. Teradata validates them and fails the write outright with
"Invalid connection parameter name path".
- Writing a pathless format now calls the no-argument save(), so the path
never enters the writer options. Only jdbc is pathless today; the set is
named so other option-addressed sinks can join it.
- The error message for those formats no longer quotes a path that played
no part in the write.
- SparkFileOutput no longer demands a file path when the format is
pathless. It previously rejected the transform before it ran, forcing
users to invent a path that was then sent to the driver as the property
causing the failure.
The lake table writer shares writeDataset but only ever passes delta or
iceberg, both path-based, so it is unaffected.
Tests capture the connection properties Spark builds via a delegating
driver and assert "path" is absent, rather than depending on a driver
strict enough to reject it - H2, like Postgres and MySQL, ignores it.
---
plugins/engines/spark/pom.xml | 15 ++
.../spark/pipeline/handler/SparkFileIoSupport.java | 27 +++-
.../pipeline/handler/SparkFileOutputHandler.java | 4 +-
.../pipeline/handler/CapturingJdbcDriver.java | 87 +++++++++++
.../pipeline/handler/SparkFileIoJdbcWriteTest.java | 168 +++++++++++++++++++++
5 files changed, 299 insertions(+), 2 deletions(-)
diff --git a/plugins/engines/spark/pom.xml b/plugins/engines/spark/pom.xml
index 8f90135bcf..f456a16483 100644
--- a/plugins/engines/spark/pom.xml
+++ b/plugins/engines/spark/pom.xml
@@ -52,6 +52,13 @@
<type>pom</type>
<scope>import</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hop</groupId>
+ <artifactId>hop-libs-jdbc</artifactId>
+ <version>${project.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
</dependencies>
</dependencyManagement>
@@ -196,6 +203,14 @@
</exclusion>
</exclusions>
</dependency>
+ <!-- Real backing database for the jdbc write tests (issue #8138). H2
ignores unknown
+ connection properties, so the tests assert what Spark hands the
driver rather than
+ relying on a driver being strict enough to complain. -->
+ <dependency>
+ <groupId>com.h2database</groupId>
+ <artifactId>h2</artifactId>
+ <scope>test</scope>
+ </dependency>
<!-- Filter Rows for target-stream unit tests -->
<dependency>
<groupId>org.apache.hop</groupId>
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkFileIoSupport.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkFileIoSupport.java
index 3c3c2f1239..5450d6d082 100644
---
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkFileIoSupport.java
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkFileIoSupport.java
@@ -19,6 +19,7 @@ package org.apache.hop.spark.pipeline.handler;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.Set;
import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.variables.IVariables;
@@ -34,6 +35,20 @@ public final class SparkFileIoSupport {
private SparkFileIoSupport() {}
+ /**
+ * Formats whose sink is not addressed by a path. Spark's {@code
DataFrameWriter.save(String)}
+ * stores its argument as the {@code path} option before saving, and the
JDBC provider forwards
+ * every option it does not recognise to the driver as a connection
property. Drivers that
+ * validate connection properties (Teradata) then reject {@code path}
outright, while Postgres and
+ * MySQL silently ignore it — which is why this only surfaces on some
databases (issue #8138).
+ */
+ private static final Set<String> PATHLESS_FORMATS = Set.of("jdbc");
+
+ /** Whether writing this format should omit the path rather than pass it to
Spark as an option. */
+ public static boolean isPathless(String format) {
+ return format != null &&
PATHLESS_FORMATS.contains(normalizeFormat(format));
+ }
+
public static String normalizeFormat(String format) {
if (StringUtils.isEmpty(format)) {
return SparkFileInputMeta.FORMAT_CSV;
@@ -94,9 +109,19 @@ public final class SparkFileIoSupport {
if (partitionColumns != null && partitionColumns.length > 0) {
writer = writer.partitionBy(partitionColumns);
}
+ boolean pathless = isPathless(format);
try {
- writer.save(path);
+ if (pathless) {
+ // Passing the path here would land in the writer options and reach
the JDBC driver as a
+ // connection property (issue #8138).
+ writer.save();
+ } else {
+ writer.save(path);
+ }
} catch (Exception e) {
+ if (pathless) {
+ throw new HopException("Error writing Spark Dataset as " + format, e);
+ }
throw new HopException(
SparkPathDialect.withPathHint(
"Error writing Spark Dataset to '" + path + "' as " + format,
path),
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkFileOutputHandler.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkFileOutputHandler.java
index 72388cb338..587456adc3 100644
---
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkFileOutputHandler.java
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkFileOutputHandler.java
@@ -90,7 +90,9 @@ public class SparkFileOutputHandler extends
SparkBaseTransformHandler {
}
String resolved = variables.resolve(meta.getFilePath());
String path = SparkPathDialect.toSparkUri(resolved, runConfiguration);
- if (StringUtils.isEmpty(path)) {
+ // A pathless sink such as jdbc is addressed entirely through its options,
so requiring a file
+ // path would only force the user to invent one that is then ignored
(issue #8138).
+ if (StringUtils.isEmpty(path) &&
!SparkFileIoSupport.isPathless(meta.getFileFormat())) {
throw new HopException(
"Spark File Output '" + transformMeta.getName() + "' has no file
path configured");
}
diff --git
a/plugins/engines/spark/src/test/java/org/apache/hop/spark/pipeline/handler/CapturingJdbcDriver.java
b/plugins/engines/spark/src/test/java/org/apache/hop/spark/pipeline/handler/CapturingJdbcDriver.java
new file mode 100644
index 0000000000..e096c992c7
--- /dev/null
+++
b/plugins/engines/spark/src/test/java/org/apache/hop/spark/pipeline/handler/CapturingJdbcDriver.java
@@ -0,0 +1,87 @@
+/*
+ * 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.hop.spark.pipeline.handler;
+
+import java.sql.Connection;
+import java.sql.Driver;
+import java.sql.DriverManager;
+import java.sql.DriverPropertyInfo;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.logging.Logger;
+
+/**
+ * Test driver that records the connection properties Spark builds and then
delegates to the real
+ * database behind the {@link #PREFIX}.
+ *
+ * <p>Used by {@link SparkFileIoJdbcWriteTest} to check directly that {@code
path} never reaches a
+ * driver (issue #8138), rather than depending on a driver being strict enough
to reject it —
+ * Teradata rejects it, but Postgres, MySQL and H2 all ignore properties they
do not recognise.
+ *
+ * <p>Top-level rather than nested on purpose: Spark's {@code DriverRegistry}
matches a registered
+ * driver by canonical name, which for a nested class does not equal the
binary name it is loaded
+ * by, and the lookup then fails with an internal error.
+ */
+public class CapturingJdbcDriver implements Driver {
+
+ public static final String PREFIX = "jdbc:capture:";
+ public static final List<Properties> CAPTURED = new CopyOnWriteArrayList<>();
+
+ @Override
+ public Connection connect(String url, Properties info) throws SQLException {
+ if (!acceptsURL(url)) {
+ return null;
+ }
+ Properties copy = new Properties();
+ copy.putAll(info);
+ CAPTURED.add(copy);
+ return DriverManager.getConnection(url.substring(PREFIX.length()), info);
+ }
+
+ @Override
+ public boolean acceptsURL(String url) {
+ return url != null && url.startsWith(PREFIX);
+ }
+
+ @Override
+ public DriverPropertyInfo[] getPropertyInfo(String url, Properties info) {
+ return new DriverPropertyInfo[0];
+ }
+
+ @Override
+ public int getMajorVersion() {
+ return 1;
+ }
+
+ @Override
+ public int getMinorVersion() {
+ return 0;
+ }
+
+ @Override
+ public boolean jdbcCompliant() {
+ return false;
+ }
+
+ @Override
+ public Logger getParentLogger() {
+ return Logger.getLogger("hop-8138-capturing-driver");
+ }
+}
diff --git
a/plugins/engines/spark/src/test/java/org/apache/hop/spark/pipeline/handler/SparkFileIoJdbcWriteTest.java
b/plugins/engines/spark/src/test/java/org/apache/hop/spark/pipeline/handler/SparkFileIoJdbcWriteTest.java
new file mode 100644
index 0000000000..9ec7f1d926
--- /dev/null
+++
b/plugins/engines/spark/src/test/java/org/apache/hop/spark/pipeline/handler/SparkFileIoJdbcWriteTest.java
@@ -0,0 +1,168 @@
+/*
+ * 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.hop.spark.pipeline.handler;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Properties;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.logging.HopLogStore;
+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.SparkSession;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Issue #8138: writing with {@code file_format=jdbc} must not hand Spark a
path.
+ *
+ * <p>{@code DataFrameWriter.save(String)} records its argument as the {@code
path} option, and the
+ * JDBC provider passes every option it does not recognise to the driver as a
connection property.
+ * Teradata validates connection properties and rejects {@code path};
Postgres, MySQL and H2 all
+ * ignore properties they do not know, which is why the bug is invisible
against them.
+ *
+ * <p>Rather than depend on a strict driver being available, these tests
assert the thing that is
+ * actually wrong: what Spark hands the driver. {@link CapturingJdbcDriver}
records the {@link
+ * Properties} of every connection and delegates to H2, so the absence of
{@code path} is checked
+ * directly instead of inferred from whether some driver happened to complain.
+ */
+class SparkFileIoJdbcWriteTest {
+
+ private static SparkSession spark;
+ private static final String H2_URL = "jdbc:h2:mem:hop8138;DB_CLOSE_DELAY=-1";
+ private static final String URL = CapturingJdbcDriver.PREFIX + H2_URL;
+
+ @BeforeAll
+ static void start() throws Exception {
+ HopEnvironment.init();
+ HopLogStore.init();
+ spark =
+ SparkSession.builder()
+ .appName("hop-spark-jdbc-write-test")
+ .master("local[2]")
+ .config("spark.ui.enabled", "false")
+ .config("spark.ui.showConsoleProgress", "false")
+ .config("spark.metrics.staticSources.enabled", "false")
+ .config("spark.sql.shuffle.partitions", "2")
+ .config("spark.driver.host", "localhost")
+ .getOrCreate();
+ DriverManager.registerDriver(new CapturingJdbcDriver());
+ // Keep the in-memory database alive for the duration of the test class.
+ DriverManager.getConnection(H2_URL).close();
+ }
+
+ @AfterAll
+ static void stop() {
+ if (spark != null) {
+ spark.stop();
+ }
+ }
+
+ private Dataset<Row> sampleRows() {
+ StructType schema =
+ DataTypes.createStructType(
+ new org.apache.spark.sql.types.StructField[] {
+ DataTypes.createStructField("id", DataTypes.IntegerType, false),
+ DataTypes.createStructField("name", DataTypes.StringType, true)
+ });
+ return spark.createDataFrame(
+ java.util.List.of(RowFactory.create(1, "alpha"), RowFactory.create(2,
"beta")), schema);
+ }
+
+ private Map<String, String> jdbcOptions(String table) {
+ Map<String, String> options = new LinkedHashMap<>();
+ options.put("url", URL);
+ options.put("dbtable", table);
+ options.put("driver", CapturingJdbcDriver.class.getName());
+ return options;
+ }
+
+ /**
+ * The reported defect: the meaningless file path the transform forces the
user to invent must not
+ * reach the driver as a connection property. Fails before the fix with
{@code path} present.
+ */
+ @Test
+ void jdbcWriteDoesNotSendThePathAsAConnectionProperty() throws Exception {
+ CapturingJdbcDriver.CAPTURED.clear();
+ SparkFileIoSupport.writeDataset(
+ sampleRows(),
+ "jdbc",
+ "/tmp/hop-bug-repro/category_sales_summary",
+ SaveMode.Overwrite,
+ jdbcOptions("SALES_SUMMARY"),
+ null,
+ null);
+
+ assertFalse(
+ CapturingJdbcDriver.CAPTURED.isEmpty(), "the driver was never asked
for a connection");
+ for (Properties p : CapturingJdbcDriver.CAPTURED) {
+ assertFalse(
+ p.containsKey("path"),
+ "Spark sent 'path' as a connection property: " +
p.stringPropertyNames());
+ }
+
+ try (Connection c = DriverManager.getConnection(H2_URL);
+ Statement st = c.createStatement();
+ ResultSet rs = st.executeQuery("select count(*) from SALES_SUMMARY")) {
+ assertTrue(rs.next());
+ assertEquals(2, rs.getInt(1));
+ }
+ }
+
+ /** The same write with no path at all must behave identically — nothing
depends on the path. */
+ @Test
+ void jdbcWriteWorksWithNoPathConfiguredAtAll() throws Exception {
+ SparkFileIoSupport.writeDataset(
+ sampleRows(), "jdbc", "", SaveMode.Overwrite, jdbcOptions("NO_PATH"),
null, null);
+
+ try (Connection c = DriverManager.getConnection(H2_URL);
+ Statement st = c.createStatement();
+ ResultSet rs = st.executeQuery("select count(*) from NO_PATH")) {
+ assertTrue(rs.next());
+ assertEquals(2, rs.getInt(1));
+ }
+ }
+
+ /** Path-based formats keep receiving their path; the change must not reach
them. */
+ @Test
+ void fileFormatsAreStillPathBased() {
+ assertTrue(SparkFileIoSupport.isPathless("jdbc"));
+ assertTrue(SparkFileIoSupport.isPathless("JDBC"));
+ assertTrue(SparkFileIoSupport.isPathless(" jdbc "));
+ assertFalse(SparkFileIoSupport.isPathless("csv"));
+ assertFalse(SparkFileIoSupport.isPathless("parquet"));
+ assertFalse(SparkFileIoSupport.isPathless("orc"));
+ assertFalse(SparkFileIoSupport.isPathless("json"));
+ assertFalse(SparkFileIoSupport.isPathless("text"));
+ assertFalse(SparkFileIoSupport.isPathless("delta"));
+ assertFalse(SparkFileIoSupport.isPathless(null));
+ }
+}
