This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 2bc864c172 [Fix][Connector-V2][JDBC] Use Oracle-typed null binding for
sparse null rows (#10768)
2bc864c172 is described below
commit 2bc864c172323309b980aceb9f43223386d75322
Author: corgy-w <[email protected]>
AuthorDate: Sun Jul 19 09:10:54 2026 +0800
[Fix][Connector-V2][JDBC] Use Oracle-typed null binding for sparse null
rows (#10768)
---
.../converter/AbstractJdbcRowConverter.java | 9 ++
.../dialect/oracle/OracleJdbcRowConverter.java | 91 +++++++++++++++
.../dialect/oracle/OracleJdbcRowConverterTest.java | 122 +++++++++++++++++++++
3 files changed, 222 insertions(+)
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/converter/AbstractJdbcRowConverter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/converter/AbstractJdbcRowConverter.java
index c2a3fb04ab..010daa8245 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/converter/AbstractJdbcRowConverter.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/converter/AbstractJdbcRowConverter.java
@@ -260,6 +260,15 @@ public abstract class AbstractJdbcRowConverter implements
JdbcRowConverter {
statement.setObject(statementIndex, null);
}
+ protected void setNullToStatementByDataType(
+ PreparedStatement statement,
+ SeaTunnelDataType<?> seaTunnelDataType,
+ int statementIndex,
+ @Nullable String sourceType)
+ throws SQLException {
+ statement.setObject(statementIndex, null);
+ }
+
protected void setValueToStatementByDataType(
Object value,
PreparedStatement statement,
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleJdbcRowConverter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleJdbcRowConverter.java
index c0a3d88653..476aa886fc 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleJdbcRowConverter.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleJdbcRowConverter.java
@@ -28,10 +28,16 @@ import java.io.ByteArrayInputStream;
import java.io.StringReader;
import java.sql.PreparedStatement;
import java.sql.SQLException;
+import java.sql.Types;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_BLOB;
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_CHAR;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_CLOB;
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_NCHAR;
import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_NCLOB;
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_NVARCHAR2;
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_VARCHAR;
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_VARCHAR2;
public class OracleJdbcRowConverter extends AbstractJdbcRowConverter {
@@ -40,6 +46,91 @@ public class OracleJdbcRowConverter extends
AbstractJdbcRowConverter {
return DatabaseIdentifier.ORACLE;
}
+ @Override
+ protected void setNullToStatementByDataType(
+ PreparedStatement statement,
+ SeaTunnelDataType<?> seaTunnelDataType,
+ int statementIndex,
+ @Nullable String sourceType)
+ throws SQLException {
+ if (ORACLE_CLOB.equals(sourceType)) {
+ statement.setNull(statementIndex, Types.CLOB);
+ return;
+ }
+ if (ORACLE_NCLOB.equals(sourceType)) {
+ statement.setNull(statementIndex, Types.NCLOB);
+ return;
+ }
+ if (ORACLE_BLOB.equals(sourceType)) {
+ statement.setNull(statementIndex, Types.BLOB);
+ return;
+ }
+ if (ORACLE_CHAR.equals(sourceType)
+ || ORACLE_NCHAR.equals(sourceType)
+ || ORACLE_VARCHAR.equals(sourceType)
+ || ORACLE_VARCHAR2.equals(sourceType)
+ || ORACLE_NVARCHAR2.equals(sourceType)) {
+ statement.setNull(statementIndex, Types.VARCHAR);
+ return;
+ }
+
+ switch (seaTunnelDataType.getSqlType()) {
+ case STRING:
+ statement.setNull(statementIndex, Types.VARCHAR);
+ break;
+ case BOOLEAN:
+ statement.setNull(statementIndex, Types.INTEGER);
+ break;
+ case TINYINT:
+ statement.setNull(statementIndex, Types.TINYINT);
+ break;
+ case SMALLINT:
+ statement.setNull(statementIndex, Types.SMALLINT);
+ break;
+ case INT:
+ statement.setNull(statementIndex, Types.INTEGER);
+ break;
+ case BIGINT:
+ statement.setNull(statementIndex, Types.BIGINT);
+ break;
+ case FLOAT:
+ statement.setNull(statementIndex, Types.FLOAT);
+ break;
+ case DOUBLE:
+ statement.setNull(statementIndex, Types.DOUBLE);
+ break;
+ case DECIMAL:
+ statement.setNull(statementIndex, Types.DECIMAL);
+ break;
+ case DATE:
+ statement.setNull(statementIndex, Types.DATE);
+ break;
+ case TIME:
+ statement.setNull(statementIndex, Types.TIME);
+ break;
+ case TIMESTAMP:
+ statement.setNull(statementIndex, Types.TIMESTAMP);
+ break;
+ case TIMESTAMP_TZ:
+ statement.setNull(statementIndex,
Types.TIMESTAMP_WITH_TIMEZONE);
+ break;
+ case BYTES:
+ statement.setNull(statementIndex, Types.BINARY);
+ break;
+ case NULL:
+ statement.setNull(statementIndex, Types.NULL);
+ break;
+ case ARRAY:
+ statement.setNull(statementIndex, Types.ARRAY);
+ break;
+ case MAP:
+ case ROW:
+ default:
+ super.setNullToStatementByDataType(
+ statement, seaTunnelDataType, statementIndex,
sourceType);
+ }
+ }
+
@Override
protected void setValueToStatementByDataType(
Object value,
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleJdbcRowConverterTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleJdbcRowConverterTest.java
new file mode 100644
index 0000000000..00fec54cac
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/oracle/OracleJdbcRowConverterTest.java
@@ -0,0 +1,122 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle;
+
+import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.ArrayType;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.AbstractJdbcRowConverter;
+
+import org.junit.jupiter.api.Test;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Types;
+import java.util.Collections;
+
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_BLOB;
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_CLOB;
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_NCLOB;
+import static
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.oracle.OracleTypeConverter.ORACLE_VARCHAR2;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+public class OracleJdbcRowConverterTest {
+
+ private static final String FIELD_NAME = "payload";
+
+ private final OracleJdbcRowConverter oracleJdbcRowConverter = new
OracleJdbcRowConverter();
+
+ @Test
+ public void testNullClobUsesTypedNullBinding() throws SQLException {
+ assertOracleNullBinding(BasicType.STRING_TYPE, ORACLE_CLOB,
Types.CLOB);
+ }
+
+ @Test
+ public void testNullNclobUsesTypedNullBinding() throws SQLException {
+ assertOracleNullBinding(BasicType.STRING_TYPE, ORACLE_NCLOB,
Types.NCLOB);
+ }
+
+ @Test
+ public void testNullBlobUsesTypedNullBinding() throws SQLException {
+ assertOracleNullBinding(ArrayType.BYTE_ARRAY_TYPE, ORACLE_BLOB,
Types.BLOB);
+ }
+
+ @Test
+ public void testNullVarchar2UsesTypedNullBinding() throws SQLException {
+ assertOracleNullBinding(BasicType.STRING_TYPE, ORACLE_VARCHAR2,
Types.VARCHAR);
+ }
+
+ @Test
+ public void testNonOracleConverterKeepsDefaultUntypedNullBinding() throws
SQLException {
+ TableSchema tableSchema = createTableSchema(BasicType.STRING_TYPE,
null);
+ TableSchema databaseTableSchema =
createTableSchema(BasicType.STRING_TYPE, ORACLE_CLOB);
+ PreparedStatement statement = mock(PreparedStatement.class);
+
+ new DefaultJdbcRowConverter()
+ .toExternal(
+ tableSchema,
+ databaseTableSchema,
+ new SeaTunnelRow(new Object[] {null}),
+ statement);
+
+ verify(statement).setObject(1, null);
+ verify(statement, never()).setNull(1, Types.CLOB);
+ verifyNoMoreInteractions(statement);
+ }
+
+ private void assertOracleNullBinding(
+ SeaTunnelDataType<?> seaTunnelDataType, String sourceType, int
expectedJdbcType)
+ throws SQLException {
+ TableSchema tableSchema = createTableSchema(seaTunnelDataType, null);
+ TableSchema databaseTableSchema = createTableSchema(seaTunnelDataType,
sourceType);
+ PreparedStatement statement = mock(PreparedStatement.class);
+
+ oracleJdbcRowConverter.toExternal(
+ tableSchema, databaseTableSchema, new SeaTunnelRow(new
Object[] {null}), statement);
+
+ verify(statement).setNull(1, expectedJdbcType);
+ verify(statement, never()).setObject(1, null);
+ verifyNoMoreInteractions(statement);
+ }
+
+ private TableSchema createTableSchema(
+ SeaTunnelDataType<?> seaTunnelDataType, String sourceType) {
+ PhysicalColumn.PhysicalColumnBuilder columnBuilder =
+
PhysicalColumn.builder().name(FIELD_NAME).dataType(seaTunnelDataType);
+ if (sourceType != null) {
+ columnBuilder.sourceType(sourceType);
+ }
+ Column column = columnBuilder.build();
+ return
TableSchema.builder().columns(Collections.singletonList(column)).build();
+ }
+
+ private static final class DefaultJdbcRowConverter extends
AbstractJdbcRowConverter {
+
+ @Override
+ public String converterName() {
+ return "default-test-converter";
+ }
+ }
+}
