och5351 commented on code in PR #188:
URL:
https://github.com/apache/flink-connector-jdbc/pull/188#discussion_r3118144708
##########
flink-connector-jdbc-postgres/src/main/java/org/apache/flink/connector/jdbc/postgres/database/dialect/PostgresDialect.java:
##########
@@ -134,4 +137,224 @@ public Set<LogicalTypeRoot> supportedTypes() {
LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE,
LogicalTypeRoot.ARRAY);
}
+
+ /**
+ * Generate UNNEST-based batch insert statement for PostgreSQL.
+ *
+ * <p>This method generates SQL that uses PostgreSQL's UNNEST() function
for bulk inserts.
+ * Instead of executing multiple INSERT statements, it creates a single
INSERT with arrays,
+ * providing significant performance improvements (5-10x) and ensuring a
single query plan
+ * regardless of batch size.
+ *
+ * <p>Example output for a table with columns (id, name, age):
+ *
+ * <pre>
+ * INSERT INTO users (id, name, age)
+ * SELECT * FROM UNNEST(?::INTEGER[], ?::VARCHAR[], ?::INTEGER[]) AS t(id,
name, age)
+ * </pre>
+ *
+ * @param tableName the target table name
+ * @param fieldNames array of field names in the order they appear in the
table
+ * @param fieldTypes array of SQL type names corresponding to each field
+ * @return Optional containing the UNNEST SQL statement
+ */
+ @Override
+ public Optional<String> getBatchInsertStatement(
+ String tableName, String[] fieldNames, String[] fieldTypes) {
+
+ if (fieldNames.length == 0) {
+ return Optional.empty();
+ }
+
+ StringBuilder sql = new StringBuilder();
+
+ // INSERT INTO table_name
+ sql.append("INSERT INTO ").append(quoteIdentifier(tableName));
+
+ // (col1, col2, col3)
+ sql.append(" (");
+ for (int i = 0; i < fieldNames.length; i++) {
+ if (i > 0) {
+ sql.append(", ");
+ }
+ sql.append(quoteIdentifier(fieldNames[i]));
+ }
+ sql.append(")");
+
+ // SELECT * FROM UNNEST
+ sql.append(" SELECT * FROM UNNEST(");
+
+ // ?::type1[], ?::type2[], ?::type3[]
+ for (int i = 0; i < fieldNames.length; i++) {
+ if (i > 0) {
+ sql.append(", ");
+ }
+ String baseType = extractBaseType(fieldTypes[i]);
+ sql.append("?::").append(baseType).append("[]");
+ }
+
+ // AS t(col1, col2, col3)
+ sql.append(") AS t(");
+ for (int i = 0; i < fieldNames.length; i++) {
+ if (i > 0) {
+ sql.append(", ");
+ }
+ sql.append(quoteIdentifier(fieldNames[i]));
+ }
+ sql.append(")");
+
+ return Optional.of(sql.toString());
+ }
+
+ /**
+ * Generate UNNEST-based batch upsert statement for PostgreSQL.
+ *
+ * <p>This extends the batch insert statement with PostgreSQL's ON
CONFLICT clause to handle
+ * upsert semantics. The statement will insert new rows and update
existing ones on conflict.
+ *
+ * <p>Example output:
+ *
+ * <pre>
+ * INSERT INTO users (id, name, age)
+ * SELECT * FROM UNNEST(?::INTEGER[], ?::VARCHAR[], ?::INTEGER[]) AS t(id,
name, age)
+ * ON CONFLICT (id) DO UPDATE SET name=EXCLUDED.name, age=EXCLUDED.age
+ * </pre>
+ *
+ * @param tableName the target table name
+ * @param fieldNames array of all field names
+ * @param fieldTypes array of SQL type names
+ * @param uniqueKeyFields array of unique key field names for conflict
detection
+ * @return Optional containing the UNNEST upsert SQL statement
+ */
+ @Override
+ public Optional<String> getBatchUpsertStatement(
+ String tableName, String[] fieldNames, String[] fieldTypes,
String[] uniqueKeyFields) {
+
+ Optional<String> batchInsert = getBatchInsertStatement(tableName,
fieldNames, fieldTypes);
+
+ if (!batchInsert.isPresent()) {
+ return Optional.empty();
+ }
+
+ StringBuilder sql = new StringBuilder(batchInsert.get());
+
+ // ON CONFLICT (key1, key2)
+ sql.append(" ON CONFLICT (");
+ for (int i = 0; i < uniqueKeyFields.length; i++) {
+ if (i > 0) {
+ sql.append(", ");
+ }
+ sql.append(quoteIdentifier(uniqueKeyFields[i]));
+ }
+
+ // Determine non-key fields for UPDATE clause
+ Set<String> keySet = new HashSet<>(Arrays.asList(uniqueKeyFields));
+ List<String> nonKeyFields =
+ Arrays.stream(fieldNames)
+ .filter(f -> !keySet.contains(f))
+ .collect(Collectors.toList());
+
+ if (nonKeyFields.isEmpty()) {
+ sql.append(") DO NOTHING");
+ } else {
+ // DO UPDATE SET col1=EXCLUDED.col1, col2=EXCLUDED.col2
+ sql.append(") DO UPDATE SET ");
+ for (int i = 0; i < nonKeyFields.size(); i++) {
+ if (i > 0) {
+ sql.append(", ");
+ }
+ String field = quoteIdentifier(nonKeyFields.get(i));
+ sql.append(field).append("=EXCLUDED.").append(field);
+ }
+ }
+
+ return Optional.of(sql.toString());
+ }
+
+ /**
+ * Extract base type name for PostgreSQL array creation.
+ *
+ * <p>PostgreSQL's createArrayOf() requires base type names without:
+ *
+ * <ul>
+ * <li>Array brackets: text[] → text
+ * <li>Length modifiers: varchar(255) → varchar
+ * <li>Precision/scale: numeric(10,2) → numeric
+ * </ul>
+ *
+ * @param typeName the full SQL type name
+ * @return the base type name suitable for createArrayOf()
+ */
+ private String extractBaseType(String typeName) {
+ // Remove array brackets: text[][] -> text
+ typeName = typeName.replaceAll("\\[\\]", "").trim();
+
+ // Remove precision/scale modifiers: varchar(255) -> varchar,
numeric(10,2) -> numeric
+ int parenIndex = typeName.indexOf('(');
+ if (parenIndex > 0) {
+ typeName = typeName.substring(0, parenIndex).trim();
+ }
+
+ return typeName.toUpperCase();
+ }
+
+ /**
+ * Get PostgreSQL type name for array operations.
+ *
+ * <p>Returns the PostgreSQL type name used for:
+ *
+ * <ul>
+ * <li>{@code Connection.createArrayOf(typeName, values)}
+ * <li>SQL type casting: {@code ?::typename[]}
+ * </ul>
+ *
+ * <p>PostgreSQL is case-insensitive for type names, but we follow
PostgreSQL's convention of
+ * using lowercase type names as shown in {@code \dT} command output.
+ *
+ * @param logicalType the Flink logical type
+ * @return PostgreSQL type name (lowercase, following PostgreSQL
convention)
+ * @throws UnsupportedOperationException if the type is not supported for
UNNEST
+ */
+ @Override
+ public String getArrayTypeName(LogicalType logicalType) {
Review Comment:
It appears that the uppercase type is actually being used. It would be best
to change it to uppercase.
> PostgresDialectTest.java
##########
flink-connector-jdbc-postgres/src/main/java/org/apache/flink/connector/jdbc/postgres/database/dialect/PostgresDialect.java:
##########
@@ -134,4 +137,224 @@ public Set<LogicalTypeRoot> supportedTypes() {
LogicalTypeRoot.TIMESTAMP_WITH_LOCAL_TIME_ZONE,
LogicalTypeRoot.ARRAY);
}
+
+ /**
+ * Generate UNNEST-based batch insert statement for PostgreSQL.
+ *
+ * <p>This method generates SQL that uses PostgreSQL's UNNEST() function
for bulk inserts.
+ * Instead of executing multiple INSERT statements, it creates a single
INSERT with arrays,
+ * providing significant performance improvements (5-10x) and ensuring a
single query plan
+ * regardless of batch size.
+ *
+ * <p>Example output for a table with columns (id, name, age):
+ *
+ * <pre>
+ * INSERT INTO users (id, name, age)
+ * SELECT * FROM UNNEST(?::INTEGER[], ?::VARCHAR[], ?::INTEGER[]) AS t(id,
name, age)
+ * </pre>
+ *
+ * @param tableName the target table name
+ * @param fieldNames array of field names in the order they appear in the
table
+ * @param fieldTypes array of SQL type names corresponding to each field
+ * @return Optional containing the UNNEST SQL statement
+ */
+ @Override
+ public Optional<String> getBatchInsertStatement(
+ String tableName, String[] fieldNames, String[] fieldTypes) {
+
+ if (fieldNames.length == 0) {
+ return Optional.empty();
+ }
+
+ StringBuilder sql = new StringBuilder();
+
+ // INSERT INTO table_name
+ sql.append("INSERT INTO ").append(quoteIdentifier(tableName));
+
+ // (col1, col2, col3)
+ sql.append(" (");
+ for (int i = 0; i < fieldNames.length; i++) {
+ if (i > 0) {
+ sql.append(", ");
+ }
+ sql.append(quoteIdentifier(fieldNames[i]));
+ }
+ sql.append(")");
+
+ // SELECT * FROM UNNEST
+ sql.append(" SELECT * FROM UNNEST(");
+
+ // ?::type1[], ?::type2[], ?::type3[]
+ for (int i = 0; i < fieldNames.length; i++) {
+ if (i > 0) {
+ sql.append(", ");
+ }
+ String baseType = extractBaseType(fieldTypes[i]);
+ sql.append("?::").append(baseType).append("[]");
+ }
+
+ // AS t(col1, col2, col3)
+ sql.append(") AS t(");
+ for (int i = 0; i < fieldNames.length; i++) {
+ if (i > 0) {
+ sql.append(", ");
+ }
+ sql.append(quoteIdentifier(fieldNames[i]));
+ }
+ sql.append(")");
+
+ return Optional.of(sql.toString());
+ }
+
+ /**
+ * Generate UNNEST-based batch upsert statement for PostgreSQL.
+ *
+ * <p>This extends the batch insert statement with PostgreSQL's ON
CONFLICT clause to handle
+ * upsert semantics. The statement will insert new rows and update
existing ones on conflict.
+ *
+ * <p>Example output:
+ *
+ * <pre>
+ * INSERT INTO users (id, name, age)
+ * SELECT * FROM UNNEST(?::INTEGER[], ?::VARCHAR[], ?::INTEGER[]) AS t(id,
name, age)
+ * ON CONFLICT (id) DO UPDATE SET name=EXCLUDED.name, age=EXCLUDED.age
+ * </pre>
+ *
+ * @param tableName the target table name
+ * @param fieldNames array of all field names
+ * @param fieldTypes array of SQL type names
+ * @param uniqueKeyFields array of unique key field names for conflict
detection
+ * @return Optional containing the UNNEST upsert SQL statement
+ */
+ @Override
+ public Optional<String> getBatchUpsertStatement(
+ String tableName, String[] fieldNames, String[] fieldTypes,
String[] uniqueKeyFields) {
+
+ Optional<String> batchInsert = getBatchInsertStatement(tableName,
fieldNames, fieldTypes);
+
+ if (!batchInsert.isPresent()) {
+ return Optional.empty();
+ }
+
+ StringBuilder sql = new StringBuilder(batchInsert.get());
+
+ // ON CONFLICT (key1, key2)
+ sql.append(" ON CONFLICT (");
+ for (int i = 0; i < uniqueKeyFields.length; i++) {
+ if (i > 0) {
+ sql.append(", ");
+ }
+ sql.append(quoteIdentifier(uniqueKeyFields[i]));
+ }
+
+ // Determine non-key fields for UPDATE clause
+ Set<String> keySet = new HashSet<>(Arrays.asList(uniqueKeyFields));
+ List<String> nonKeyFields =
+ Arrays.stream(fieldNames)
+ .filter(f -> !keySet.contains(f))
+ .collect(Collectors.toList());
+
+ if (nonKeyFields.isEmpty()) {
+ sql.append(") DO NOTHING");
+ } else {
+ // DO UPDATE SET col1=EXCLUDED.col1, col2=EXCLUDED.col2
+ sql.append(") DO UPDATE SET ");
+ for (int i = 0; i < nonKeyFields.size(); i++) {
+ if (i > 0) {
+ sql.append(", ");
+ }
+ String field = quoteIdentifier(nonKeyFields.get(i));
+ sql.append(field).append("=EXCLUDED.").append(field);
+ }
+ }
+
+ return Optional.of(sql.toString());
+ }
+
+ /**
+ * Extract base type name for PostgreSQL array creation.
+ *
+ * <p>PostgreSQL's createArrayOf() requires base type names without:
+ *
+ * <ul>
+ * <li>Array brackets: text[] → text
+ * <li>Length modifiers: varchar(255) → varchar
+ * <li>Precision/scale: numeric(10,2) → numeric
+ * </ul>
+ *
+ * @param typeName the full SQL type name
+ * @return the base type name suitable for createArrayOf()
+ */
+ private String extractBaseType(String typeName) {
+ // Remove array brackets: text[][] -> text
+ typeName = typeName.replaceAll("\\[\\]", "").trim();
+
+ // Remove precision/scale modifiers: varchar(255) -> varchar,
numeric(10,2) -> numeric
+ int parenIndex = typeName.indexOf('(');
+ if (parenIndex > 0) {
+ typeName = typeName.substring(0, parenIndex).trim();
+ }
+
+ return typeName.toUpperCase();
Review Comment:
If you use uppercase for getArrayTypeName, the toUpperCase() method seems
unnecessary.
--
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]
