Create INSERT Statements as Target Output

import de.ubs.xdm.utils.modification.ModificationContext

// Class-level variables
String insertPrefix            // "INSERT INTO schema.table (COL1, COL2, ...) VALUES ("
List<String> columnNames       // Column names in column order
List<String> columnTypesLocal  // Column types in column order
File outputFile                // Output file for the INSERT statements

/**
 * Initialization phase of the Modification Method.
 */
def init() {
    // 1. Retrieve table information.
    // getCurrentTableSchema() returns the schema, getCurrentTableName() returns the table name.
    String schemaName = ctx.getCurrentTableSchema()
    String tableName  = ctx.getCurrentTableName()

    String fullTableName = (schemaName != null && !schemaName.trim().isEmpty())
            ? "${schemaName}.${tableName}"
            : tableName

    // 2. Retrieve column information.
    def cols = ctx.getDataColumns()

    // Sort columns by column position if required.
    // cols = cols.sort { it.getColumnNo() }

    columnNames = cols.collect { def c ->
        c.getName()
    }

    columnTypesLocal = cols.collect { def c ->
        String typeName = c.getTypeName()
        return typeName != null ? typeName.toUpperCase() : null
    }

    // 3. Build the INSERT statement prefix.
    String columnList = columnNames.collect { "\"${it}\"" }.join(", ")
    insertPrefix = "INSERT INTO ${fullTableName} (${columnList}) VALUES ("

    // 4. Retrieve root_dir from the task runtime properties.
    String rootDirPath = ctx.getTaskRuntimeProperty("root_dir")

    if (!rootDirPath || rootDirPath.trim().isEmpty()) {
        throw new IllegalStateException("Task runtime property 'root_dir' is not set.")
    }

    File rootDir = new File(rootDirPath)

    if (!rootDir.exists()) {
        rootDir.mkdirs()
    }

    // 5. Prepare the output file for this table.
    String safeTableName = fullTableName.replace('.', '_')
    outputFile = new File(rootDir, "insert_${safeTableName}.sql")

    if (outputFile.exists()) {
        outputFile.delete()
    }

    outputFile.createNewFile()

    return null
}

/**
 * Apply phase.
 * This method is executed once for each row.
 */
def apply() {
    List<String> values = []

    for (int i = 0; i < data.length; i++) {
        Object value = data[i]
        String type = (i < columnTypesLocal.size()) ? columnTypesLocal[i] : null

        if (value == null) {
            values << "NULL"
            continue
        }

        String upperType = type != null ? type.toUpperCase() : ""

        switch (upperType) {
            case ["CHAR", "VARCHAR", "CLOB", "GRAPHIC", "VARGRAPHIC"]:
            case { it.startsWith("CHAR(") || it.startsWith("VARCHAR(") || it.contains("CHAR") }:
                // Character-based values are written as SQL string literals.
                values << toStringLiteral(value.toString())
                break

            case "DATE":
                // Expected format: YYYY-MM-DD.
                // Add formatting here if the input value uses a different format.
                String dateString = value.toString()
                values << "DATE '${dateString}'"
                break

            case ["TIMESTAMP", "TIMESTMP"]:
                // Expected Db2 format: YYYY-MM-DD-HH.MM.SS[.ffffff].
                String timestampString = value.toString()
                values << "TIMESTAMP '${timestampString}'"
                break

            default:
                // Numeric and other values are written without additional formatting.
                values << value.toString()
        }
    }

    String valueList = values.join(", ")
    String statement = "${insertPrefix}${valueList});"

    // Append the statement to the output file.
    outputFile << (statement + System.lineSeparator())

    // Do not pass the record to further task processing.
    return false

    // Use this if the record should continue in the task processing chain.
    // return true
}

/**
 * Close phase of the Modification Method.
 */
def close() {
}

/**
 * Converts a string value into a Db2 SQL string literal.
 */
private String toStringLiteral(String value) {
    String escapedValue = value.replace("'", "''")
    return "'${escapedValue}'"
}

import java.nio.file.*

/**
* Task Stage Hook (POST Stage 2)
*
* Expected input:
* - properties["root_dir"]: Directory that contains the insert_*.sql files.
* - taskStageHook.StatementOutputFile: File parameter of the hook.
*   - taskStageHook.StatementOutputFile.content is populated with the complete SQL text.
*
* The content of all insert_*.sql files is collected in an internal string variable
* and written to taskStageHook.StatementOutputFile.content at the end.
*/

// 1. Determine root_dir.
String rootDirPath = properties?.get("root_dir") as String

if (!rootDirPath || rootDirPath.trim().isEmpty()) {
throw new IllegalStateException("Runtime property 'root_dir' is not set.")
}

File rootDir = new File(rootDirPath)

if (!rootDir.exists() || !rootDir.isDirectory()) {
throw new IllegalStateException("root_dir '${rootDirPath}' does not exist or is not a directory.")
}

// 2. Collect insert_*.sql files.
List<File> insertFiles = rootDir.listFiles(new FilenameFilter() {
@Override
boolean accept(File dir, String name) {
return name.startsWith("insert_") && name.endsWith(".sql")
}
})?.toList() ?: []

// 3. Define the processing order.
// Default option: sort alphabetically by file name.
// insertFiles.sort { it.name }

// Alternative option: use an explicit table order if required.
List<String> tableOrder = [
"production.employees",
"production.titles",
"production.salaries"
]

List<File> orderedFiles = []

tableOrder.each { String tableName ->
String fileName = "insert_${tableName.replace('.', '_')}.sql"
File file = insertFiles.find { it.name == fileName }

    if (file) {
        orderedFiles << file
    }
}

orderedFiles.addAll(insertFiles - orderedFiles)
insertFiles = orderedFiles

// 4. Collect the file contents in memory.
StringBuilder contentBuilder = new StringBuilder()

insertFiles.each { File file ->
contentBuilder.append("-- BEGIN ").append(file.name).append(System.lineSeparator())
contentBuilder.append(file.getText("UTF-8"))
contentBuilder.append(System.lineSeparator())
contentBuilder.append("-- END ").append(file.name).append(System.lineSeparator())
}

// 5. Write the collected content to the XDM file.
// The hook parameter is a File parameter, and the content is assigned as a string.
String fullContent = contentBuilder.toString()

if (taskStageHook?.StatementOutputFile == null) {
throw new IllegalStateException("Hook parameter 'StatementOutputFile' (File) is not configured.")
}

taskStageHook.StatementOutputFile.content = fullContent

// Optional: store the output size in properties if the calling workflow needs it.
// properties.put("INSERT_OUTPUT_SIZE", fullContent.length().toString())

"emp_no" < 10017

-- BEGIN insert_production_employees.sql
INSERT INTO production.employees ("emp_no", "first_name", "last_name")
VALUES (10001, 'Georgi', 'Facello');
-- END insert_production_employees.sql
-- BEGIN insert_production_titles.sql
INSERT INTO production.titles ("emp_no", "title", "from_date", "to_date")
VALUES (10001, 'Senior Engineer', DATE '2006-06-26', DATE '9999-01-01');
-- END insert_production_titles.sql