[improve][io] support array schema for JDBC postgres connector (apache#24549)

Co-authored-by: Omri Fried <100372327+omri-cavnue@users.noreply.github.com>

Author: 2 people

pulsar-io/jdbc/clickhouse/src/main/java/org/apache/pulsar/io/jdbc/ClickHouseJdbcAutoSchemaSink.java

@@ -18,15 +18,51 @@
  */
 package org.apache.pulsar.io.jdbc;
 
+import java.sql.PreparedStatement;
 import org.apache.pulsar.io.core.annotations.Connector;
 import org.apache.pulsar.io.core.annotations.IOType;
 
 @Connector(
-    name = "jdbc-clickhouse",
-    type = IOType.SINK,
-    help = "A simple JDBC sink for ClickHouse that writes pulsar messages to a database table",
-    configClass = JdbcSinkConfig.class
+        name = "jdbc-clickhouse",
+        type = IOType.SINK,
+        help = "A simple JDBC sink for ClickHouse that writes pulsar messages to a database table",
+        configClass = JdbcSinkConfig.class
 )
 public class ClickHouseJdbcAutoSchemaSink extends BaseJdbcAutoSchemaSink {
 
+    /**
+     * ClickHouse array support is not currently implemented.
+     * <p>
+     * While ClickHouse has native support for array types (e.g., Array(Int32), Array(String)),
+     * the automatic conversion from Avro arrays to ClickHouse arrays is not yet implemented
+     * in this JDBC sink. ClickHouse arrays have specific syntax and behavior that would
+     * require dedicated implementation.
+     * </p>
+     * <p>
+     * <strong>Alternatives:</strong>
+     * <ul>
+     * <li>Use PostgreSQL JDBC sink for native array support</li>
+     * <li>Implement custom ClickHouse array conversion logic</li>
+     * <li>Serialize arrays to JSON strings for storage in String columns</li>
+     * <li>Use separate tables for one-to-many relationships</li>
+     * </ul>
+     * </p>
+     * <p>
+     * <strong>Future Enhancement:</strong>
+     * This method could be enhanced to support ClickHouse-specific array conversion
+     * using the ClickHouse JDBC driver's array handling capabilities.
+     * </p>
+     *
+     * @param statement     the PreparedStatement (not used)
+     * @param index         the parameter index (not used)
+     * @param arrayValue    the array value (not used)
+     * @param targetSqlType the target SQL type (not used)
+     * @throws UnsupportedOperationException always thrown as ClickHouse array support is not implemented
+     */
+    @Override
+    protected void handleArrayValue(PreparedStatement statement, int index, Object arrayValue, String targetSqlType)
+            throws Exception {
+        throw new UnsupportedOperationException("Array types are not supported by ClickHouse JDBC sink. "
+                + "Consider using PostgreSQL JDBC sink for array support.");
+    }
 }

pulsar-io/jdbc/core/src/main/java/org/apache/pulsar/io/jdbc/BaseJdbcAutoSchemaSink.java

@@ -30,6 +30,7 @@
 import java.util.function.Function;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
 import org.apache.pulsar.client.api.schema.GenericObject;
 import org.apache.pulsar.client.api.schema.GenericRecord;
 import org.apache.pulsar.client.api.schema.KeyValueSchema;
@@ -55,6 +56,61 @@ public List<ColumnId> getColumnsForUpsert() {
         throw new IllegalStateException("UPSERT not supported");
     }
 
+    /**
+     * Handles array value binding for database-specific array types.
+     * <p>
+     * This method is called when an array value needs to be bound to a PreparedStatement
+     * parameter. Implementations should convert the array value to the appropriate
+     * database-specific array type and bind it to the statement using the appropriate
+     * JDBC method (typically {@code PreparedStatement.setArray()}).
+     * </p>
+     * <p>
+     * The method is invoked automatically by {@link #setColumnValue(PreparedStatement, int, Object, String)}
+     * when it detects an array type (specifically {@code org.apache.avro.generic.GenericData$Array}).
+     * </p>
+     * <p>
+     * <strong>Implementation Guidelines:</strong>
+     * <ul>
+     * <li>Handle null arrays by calling {@code statement.setNull(index, java.sql.Types.ARRAY)}</li>
+     * <li>Convert array elements to the appropriate database-specific types</li>
+     * <li>Use the targetSqlType parameter to determine the correct array element type</li>
+     * <li>Provide descriptive error messages for type mismatches and unsupported types</li>
+     * <li>Wrap JDBC exceptions with contextual information</li>
+     * </ul>
+     * </p>
+     * <p>
+     * <strong>Example Usage:</strong>
+     * <pre>{@code
+     * // For PostgreSQL implementation:
+     * if (arrayValue == null) {
+     *     statement.setNull(index, java.sql.Types.ARRAY);
+     *     return;
+     * }
+     *
+     * Object[] elements = convertToObjectArray(arrayValue);
+     * String postgresType = mapToPostgresType(targetSqlType);
+     * Array pgArray = connection.createArrayOf(postgresType, elements);
+     * statement.setArray(index, pgArray);
+     * }</pre>
+     * </p>
+     *
+     * @param statement     the PreparedStatement to bind the array value to
+     * @param index         the parameter index (1-based) in the PreparedStatement
+     * @param arrayValue    the array value to be bound, typically a {@code GenericData.Array} or {@code Object[]}
+     * @param targetSqlType the target SQL type name for the array column (e.g., "integer", "text", "_int4")
+     * @throws Exception if array conversion or binding fails, including:
+     *                   <ul>
+     *                   <li>{@code IllegalArgumentException} for unsupported array types or type mismatches</li>
+     *                   <li>{@code SQLException} for JDBC array creation or binding failures</li>
+     *                   <li>{@code UnsupportedOperationException} for databases that don't support arrays</li>
+     *                   </ul>
+     * @see #setColumnValue(PreparedStatement, int, Object, String)
+     * @see java.sql.PreparedStatement#setArray(int, java.sql.Array)
+     * @see java.sql.Connection#createArrayOf(String, Object[])
+     */
+    protected abstract void handleArrayValue(PreparedStatement statement, int index, Object arrayValue,
+                                             String targetSqlType) throws Exception;
+
     @Override
     public void bindValue(PreparedStatement statement, Mutation mutation) throws Exception {
         final List<ColumnId> columns = new ArrayList<>();
@@ -78,13 +134,14 @@ public void bindValue(PreparedStatement statement, Mutation mutation) throws Exc
         for (ColumnId columnId : columns) {
             String colName = columnId.getName();
             int colType = columnId.getType();
+            String typeName = columnId.getTypeName();
             if (log.isDebugEnabled()) {
                 log.debug("getting value for column: {} type: {}", colName, colType);
             }
             try {
                 Object obj = mutation.getValues().apply(colName);
                 if (obj != null) {
-                    setColumnValue(statement, index++, obj);
+                    setColumnValue(statement, index++, obj, typeName);
                 } else {
                     if (log.isDebugEnabled()) {
                         log.debug("Column {} is null", colName);
@@ -174,10 +231,73 @@ private static void setColumnNull(PreparedStatement statement, int index, int ty
 
     }
 
-    protected void setColumnValue(PreparedStatement statement, int index, Object value) throws Exception {
+    /**
+     * Sets a column value in a PreparedStatement, handling various data types including arrays.
+     * <p>
+     * This method automatically detects the type of the value and calls the appropriate
+     * PreparedStatement setter method. It supports primitive types, strings, and arrays.
+     * Array support is implemented through the {@link #handleArrayValue(PreparedStatement, int, Object, String)}
+     * method, which delegates to database-specific implementations.
+     * </p>
+     * <p>
+     * <strong>Supported Types:</strong>
+     * <ul>
+     * <li>Primitive types: Integer, Long, Double, Float, Boolean, Short</li>
+     * <li>String types: String</li>
+     * <li>Binary types: ByteString</li>
+     * <li>JSON types: GenericJsonRecord</li>
+     * <li>Array types: GenericData.Array (requires targetSqlType parameter)</li>
+     * </ul>
+     * </p>
+     * <p>
+     * <strong>Array Handling:</strong>
+     * When an array is detected (GenericData.Array), this method calls the abstract
+     * {@link #handleArrayValue(PreparedStatement, int, Object, String)} method, which
+     * must be implemented by database-specific subclasses. The targetSqlType parameter
+     * is essential for proper array type conversion.
+     * </p>
+     * <p>
+     * <strong>Example Usage:</strong>
+     * <pre>{@code
+     * // Setting a primitive value
+     * setColumnValue(statement, 1, 42, "integer");
+     *
+     * // Setting an array value (requires database-specific implementation)
+     * GenericData.Array<Integer> intArray = ...;
+     * setColumnValue(statement, 2, intArray, "integer");
+     * }</pre>
+     * </p>
+     *
+     * @param statement     the PreparedStatement to bind the value to
+     * @param index         the parameter index (1-based) in the PreparedStatement
+     * @param value         the value to be bound (null values are handled automatically)
+     * @param targetSqlType the target SQL type name for the column, required for array types
+     * @throws Exception if value binding fails, including:
+     *                   <ul>
+     *                   <li>Unsupported value types</li>
+     *                   <li>Array conversion failures (delegated to handleArrayValue)</li>
+     *                   <li>JDBC binding errors</li>
+     *                   </ul>
+     * @see #handleArrayValue(PreparedStatement, int, Object, String)
+     * @see #setColumnValue(PreparedStatement, int, Object)
+     */
+    protected void setColumnValue(PreparedStatement statement, int index, Object value,
+                                  String targetSqlType) throws Exception {
 
         log.debug("Setting column value, statement: {}, index: {}, value: {}", statement, index, value);
 
+        // Handle null values first
+        if (value == null) {
+            setColumnNull(statement, index, java.sql.Types.NULL);
+            return;
+        }
+
+        // Check for array types first, before other type checks
+        if (value instanceof GenericData.Array || value instanceof Object[]) {
+            handleArrayValue(statement, index, value, targetSqlType);
+            return;
+        }
+
         if (value instanceof Integer) {
             statement.setInt(index, (Integer) value);
         } else if (value instanceof Long) {
@@ -201,6 +321,24 @@ protected void setColumnValue(PreparedStatement statement, int index, Object val
         }
     }
 
+    /**
+     * Backward compatibility method for setColumnValue without targetSqlType parameter.
+     * This method is provided for compatibility with existing code that may call setColumnValue
+     * without the targetSqlType parameter. Arrays will not be supported when using this method.
+     *
+     * @param statement the PreparedStatement to bind the value to
+     * @param index     the parameter index (1-based) in the PreparedStatement
+     * @param value     the value to be bound
+     * @throws Exception if value binding fails or if an array is encountered (arrays require targetSqlType)
+     */
+    protected void setColumnValue(PreparedStatement statement, int index, Object value) throws Exception {
+        if (value instanceof GenericData.Array) {
+            throw new Exception("Array values require targetSqlType parameter. "
+                    + "Use setColumnValue(statement, index, value, targetSqlType) instead.");
+        }
+        setColumnValue(statement, index, value, null);
+    }
+
     private static Object getValueFromJsonNode(final JsonNode fn) {
         if (fn == null || fn.isNull()) {
             return null;
@@ -226,8 +364,8 @@ private static Object getValueFromJsonNode(final JsonNode fn) {
     }
 
     private static void fillKeyValueSchemaData(org.apache.pulsar.client.api.Schema<GenericObject> schema,
-                                        GenericObject record,
-                                        Map<String, Object> data) {
+                                               GenericObject record,
+                                               Map<String, Object> data) {
         if (record == null) {
             return;
         }
@@ -256,6 +394,51 @@ private static void fillKeyValueSchemaData(org.apache.pulsar.client.api.Schema<G
         }
     }
 
+    /**
+     * Converts an Avro field value to a Java object suitable for JDBC binding.
+     * <p>
+     * This method handles the conversion of various Avro schema types to their corresponding
+     * Java representations. It supports primitive types, strings, unions, and arrays.
+     * Array conversion is performed recursively, processing each array element according
+     * to the array's element schema.
+     * </p>
+     * <p>
+     * <strong>Supported Avro Types:</strong>
+     * <ul>
+     * <li>Primitive types: NULL, INT, LONG, DOUBLE, FLOAT, BOOLEAN</li>
+     * <li>String types: STRING, ENUM</li>
+     * <li>Union types: Automatically selects the non-null type from the union</li>
+     * <li>Array types: Recursively converts array elements to Object[]</li>
+     * </ul>
+     * </p>
+     * <p>
+     * <strong>Array Conversion:</strong>
+     * Arrays are converted by recursively processing each element according to the
+     * array's element schema. The result is an Object[] that can be further processed
+     * by database-specific array handling methods.
+     * </p>
+     * <p>
+     * <strong>Example Usage:</strong>
+     * <pre>{@code
+     * // Convert a simple integer field
+     * Schema intSchema = Schema.create(Schema.Type.INT);
+     * Object result = convertAvroField(42, intSchema); // Returns Integer(42)
+     *
+     * // Convert an array field
+     * Schema arraySchema = Schema.createArray(Schema.create(Schema.Type.STRING));
+     * GenericData.Array<String> avroArray = ...;
+     * Object result = convertAvroField(avroArray, arraySchema); // Returns String[]
+     * }</pre>
+     * </p>
+     *
+     * @param avroValue the Avro value to convert (may be null)
+     * @param schema    the Avro schema describing the value's type
+     * @return the converted Java object, or null if avroValue is null
+     * @throws IllegalArgumentException      if the avroValue doesn't match the expected schema type
+     * @throws UnsupportedOperationException if the schema type is not supported
+     * @see org.apache.avro.Schema.Type
+     * @see org.apache.avro.generic.GenericData.Array
+     */
     @VisibleForTesting
     static Object convertAvroField(Object avroValue, Schema schema) {
         if (avroValue == null) {
@@ -281,6 +464,21 @@ static Object convertAvroField(Object avroValue, Schema schema) {
                 }
                 throw new IllegalArgumentException("Found UNION schema but it doesn't contain any type");
             case ARRAY:
+                // Handle array conversion by recursively processing array elements
+                if (avroValue instanceof GenericData.Array) {
+                    GenericData.Array<?> avroArray = (GenericData.Array<?>) avroValue;
+                    Schema elementSchema = schema.getElementType();
+                    Object[] convertedArray = new Object[avroArray.size()];
+
+                    for (int i = 0; i < avroArray.size(); i++) {
+                        convertedArray[i] = convertAvroField(avroArray.get(i), elementSchema);
+                    }
+
+                    return convertedArray;
+                } else {
+                    throw new IllegalArgumentException("Expected GenericData.Array for ARRAY schema type, got: "
+                            + avroValue.getClass().getName());
+                }
             case BYTES:
             case FIXED:
             case RECORD: