Class PutDatabaseRecord
java.lang.Object
org.apache.nifi.components.AbstractConfigurableComponent
org.apache.nifi.processor.AbstractSessionFactoryProcessor
org.apache.nifi.processor.AbstractProcessor
org.apache.nifi.processors.standard.PutDatabaseRecord
- All Implemented Interfaces:
org.apache.nifi.components.ConfigurableComponent,org.apache.nifi.processor.Processor
@InputRequirement(INPUT_REQUIRED)
@Tags({"sql","record","jdbc","put","database","update","insert","delete"})
@CapabilityDescription("The PutDatabaseRecord processor uses a specified RecordReader to input (possibly multiple) records from an incoming flow file. These records are translated to SQL statements and executed as a single transaction. If any errors occur, the flow file is routed to failure or retry, and if the records are transmitted successfully, the incoming flow file is routed to success. The type of statement executed by the processor is specified via the Statement Type property, which accepts some hard-coded values such as INSERT, UPDATE, and DELETE, as well as \'Use statement.type Attribute\', which causes the processor to get the statement type from a flow file attribute. IMPORTANT: If the Statement Type is UPDATE, then the incoming records must not alter the value(s) of the primary keys (or user-specified Update Keys). If such records are encountered, the UPDATE statement issued to the database may do nothing (if no existing records with the new primary key values are found), or could inadvertently corrupt the existing data (by changing records for which the new values of the primary keys exist).")
@ReadsAttribute(attribute="statement.type",
description="If \'Use statement.type Attribute\' is selected for the Statement Type property, the value of this attribute will be used to determine the type of statement (INSERT, UPDATE, DELETE, SQL, etc.) to generate and execute.")
@WritesAttribute(attribute="putdatabaserecord.error",
description="If an error occurs during processing, the flow file will be routed to failure or retry, and this attribute will be populated with the cause of the error.")
@UseCase(description="Insert records into a database")
public class PutDatabaseRecord
extends org.apache.nifi.processor.AbstractProcessor
-
Nested Class Summary
Nested ClassesModifier and TypeClassDescription(package private) static class(package private) static classprivate static class(package private) static class(package private) static classA holder class for a SQL prepared statement and a BitSet indicating which columns are being updated (to determine which values from the record to set on the statement) A value of null for getIncludedColumns indicates that all columns/fields should be included. -
Field Summary
FieldsModifier and TypeFieldDescription(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.AllowableValue(package private) static final org.apache.nifi.components.AllowableValue(package private) static final org.apache.nifi.components.AllowableValue(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptorprivate DatabaseDialectServiceprivate RecordPath(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptorstatic final String(package private) static final org.apache.nifi.components.AllowableValue(package private) static final org.apache.nifi.components.AllowableValue(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.AllowableValue(package private) static final org.apache.nifi.components.AllowableValuestatic final Stringstatic final String(package private) static final org.apache.nifi.components.PropertyDescriptorprotected static final List<org.apache.nifi.components.PropertyDescriptor> (package private) static final String(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.processor.Relationship(package private) static final org.apache.nifi.processor.Relationshipstatic final org.apache.nifi.processor.Relationshipprivate static final Set<org.apache.nifi.processor.Relationship> (package private) static final org.apache.nifi.components.PropertyDescriptorprivate com.github.benmanes.caffeine.cache.Cache<PutDatabaseRecord.SchemaKey, TableSchema> static final String(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final String(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptorstatic final org.apache.nifi.components.PropertyDescriptorstatic final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptor(package private) static final org.apache.nifi.components.PropertyDescriptorstatic final Stringstatic final Stringstatic final Stringstatic final String(package private) static final org.apache.nifi.components.AllowableValue -
Constructor Summary
Constructors -
Method Summary
Modifier and TypeMethodDescriptionprivate voidcheckValuesForRequiredColumns(RecordSchema recordSchema, TableSchema tableSchema, PutDatabaseRecord.DMLSettings settings, NameNormalizer normalizer) private voidcloseConnection(Connection connection, boolean originalAutoCommit) protected Collection<org.apache.nifi.components.ValidationResult> customValidate(org.apache.nifi.components.ValidationContext validationContext) private voidexecuteDML(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, org.apache.nifi.flowfile.FlowFile flowFile, Connection con, RecordReader recordReader, String explicitStatementType, PutDatabaseRecord.DMLSettings settings) private voidexecuteSQL(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, org.apache.nifi.flowfile.FlowFile flowFile, Connection connection, RecordReader recordReader) (package private) PutDatabaseRecord.SqlAndIncludedColumnsgenerateDelete(RecordSchema recordSchema, String tableName, String deleteKeys, TableSchema tableSchema, PutDatabaseRecord.DMLSettings settings, NameNormalizer normalizer) (package private) PutDatabaseRecord.SqlAndIncludedColumnsgenerateInsert(RecordSchema recordSchema, String tableName, TableSchema tableSchema, PutDatabaseRecord.DMLSettings settings, NameNormalizer normalizer) (package private) StringgenerateTableName(PutDatabaseRecord.DMLSettings settings, String catalog, String schemaName, String tableName, TableSchema tableSchema) (package private) PutDatabaseRecord.SqlAndIncludedColumnsgenerateUpdate(RecordSchema recordSchema, String tableName, String updateKeys, TableSchema tableSchema, PutDatabaseRecord.DMLSettings settings, NameNormalizer normalizer) private ColumnDefinitiongetColumnDefinition(ColumnDescription columnDescription, Set<String> keyColumnNames, String columnName) getDataRecords(Record outerRecord) private StringgetJdbcUrl(Connection connection) getNormalizedColumnNames(RecordSchema schema, boolean translateFieldNames, NameNormalizer normalizer) private intgetParameterCount(String sql) Set<org.apache.nifi.processor.Relationship> getSqlStatement(StatementType statementType, RecordSchema recordSchema, String qualifiedTableName, String updateKeys, TableSchema tableSchema, PutDatabaseRecord.DMLSettings settings, NameNormalizer normalizer) private StringgetStatementType(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.flowfile.FlowFile flowFile) protected List<org.apache.nifi.components.PropertyDescriptor> getUpdateKeyColumnNames(String tableName, String updateKeys, TableSchema tableSchema) private booleanisMaxBatchSizeHardcodedToZero(org.apache.nifi.components.ValidationContext validationContext) private booleanisSupportsBatchUpdates(Connection connection) normalizeKeyColumnNamesAndCheckForValues(RecordSchema recordSchema, String updateKeys, PutDatabaseRecord.DMLSettings settings, Set<String> updateKeyColumnNames, NameNormalizer normalizer) voidonScheduled(org.apache.nifi.processor.ProcessContext context) voidonTrigger(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session) private voidputToDatabase(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, org.apache.nifi.flowfile.FlowFile flowFile, Connection connection) private voidrollbackConnection(Connection connection) private voidrouteOnException(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, Connection connection, Exception e, org.apache.nifi.flowfile.FlowFile flowFile) private voidsetParameter(PreparedStatement ps, int index, Object value, int fieldSqlType, int sqlType) private StringvalidateStatementType(String statementType, org.apache.nifi.flowfile.FlowFile flowFile) Methods inherited from class org.apache.nifi.processor.AbstractProcessor
onTriggerMethods inherited from class org.apache.nifi.processor.AbstractSessionFactoryProcessor
getControllerServiceLookup, getIdentifier, getLogger, getNodeTypeProvider, init, initialize, isConfigurationRestored, isScheduled, toString, updateConfiguredRestoredTrue, updateScheduledFalse, updateScheduledTrueMethods inherited from class org.apache.nifi.components.AbstractConfigurableComponent
equals, getPropertyDescriptor, getPropertyDescriptors, getSupportedDynamicPropertyDescriptor, hashCode, onPropertyModified, validateMethods inherited from class java.lang.Object
clone, finalize, getClass, notify, notifyAll, wait, wait, waitMethods inherited from interface org.apache.nifi.components.ConfigurableComponent
getPropertyDescriptor, getPropertyDescriptors, onPropertyModified, validateMethods inherited from interface org.apache.nifi.processor.Processor
isStateful, migrateProperties, migrateRelationships
-
Field Details
-
UPDATE_TYPE
- See Also:
-
INSERT_TYPE
- See Also:
-
DELETE_TYPE
- See Also:
-
UPSERT_TYPE
- See Also:
-
INSERT_IGNORE_TYPE
- See Also:
-
SQL_TYPE
- See Also:
-
USE_ATTR_TYPE
- See Also:
-
USE_RECORD_PATH
- See Also:
-
STATEMENT_TYPE_ATTRIBUTE
- See Also:
-
PUT_DATABASE_RECORD_ERROR
- See Also:
-
IGNORE_UNMATCHED_FIELD
static final org.apache.nifi.components.AllowableValue IGNORE_UNMATCHED_FIELD -
FAIL_UNMATCHED_FIELD
static final org.apache.nifi.components.AllowableValue FAIL_UNMATCHED_FIELD -
IGNORE_UNMATCHED_COLUMN
static final org.apache.nifi.components.AllowableValue IGNORE_UNMATCHED_COLUMN -
WARNING_UNMATCHED_COLUMN
static final org.apache.nifi.components.AllowableValue WARNING_UNMATCHED_COLUMN -
FAIL_UNMATCHED_COLUMN
static final org.apache.nifi.components.AllowableValue FAIL_UNMATCHED_COLUMN -
REL_SUCCESS
public static final org.apache.nifi.processor.Relationship REL_SUCCESS -
REL_RETRY
static final org.apache.nifi.processor.Relationship REL_RETRY -
REL_FAILURE
static final org.apache.nifi.processor.Relationship REL_FAILURE -
RELATIONSHIPS
-
RECORD_READER_FACTORY
static final org.apache.nifi.components.PropertyDescriptor RECORD_READER_FACTORY -
STATEMENT_TYPE
static final org.apache.nifi.components.PropertyDescriptor STATEMENT_TYPE -
STATEMENT_TYPE_RECORD_PATH
static final org.apache.nifi.components.PropertyDescriptor STATEMENT_TYPE_RECORD_PATH -
DATA_RECORD_PATH
static final org.apache.nifi.components.PropertyDescriptor DATA_RECORD_PATH -
DBCP_SERVICE
static final org.apache.nifi.components.PropertyDescriptor DBCP_SERVICE -
CATALOG_NAME
static final org.apache.nifi.components.PropertyDescriptor CATALOG_NAME -
SCHEMA_NAME
static final org.apache.nifi.components.PropertyDescriptor SCHEMA_NAME -
TABLE_NAME
static final org.apache.nifi.components.PropertyDescriptor TABLE_NAME -
BINARY_STRING_FORMAT_UTF8
static final org.apache.nifi.components.AllowableValue BINARY_STRING_FORMAT_UTF8 -
BINARY_STRING_FORMAT_HEXADECIMAL
static final org.apache.nifi.components.AllowableValue BINARY_STRING_FORMAT_HEXADECIMAL -
BINARY_STRING_FORMAT_BASE64
static final org.apache.nifi.components.AllowableValue BINARY_STRING_FORMAT_BASE64 -
BINARY_STRING_FORMAT
static final org.apache.nifi.components.PropertyDescriptor BINARY_STRING_FORMAT -
TRANSLATE_FIELD_NAMES
static final org.apache.nifi.components.PropertyDescriptor TRANSLATE_FIELD_NAMES -
TRANSLATION_STRATEGY
public static final org.apache.nifi.components.PropertyDescriptor TRANSLATION_STRATEGY -
TRANSLATION_PATTERN
public static final org.apache.nifi.components.PropertyDescriptor TRANSLATION_PATTERN -
UNMATCHED_FIELD_BEHAVIOR
static final org.apache.nifi.components.PropertyDescriptor UNMATCHED_FIELD_BEHAVIOR -
UNMATCHED_COLUMN_BEHAVIOR
static final org.apache.nifi.components.PropertyDescriptor UNMATCHED_COLUMN_BEHAVIOR -
UPDATE_KEYS
static final org.apache.nifi.components.PropertyDescriptor UPDATE_KEYS -
DELETE_KEYS
static final org.apache.nifi.components.PropertyDescriptor DELETE_KEYS -
FIELD_CONTAINING_SQL
static final org.apache.nifi.components.PropertyDescriptor FIELD_CONTAINING_SQL -
ALLOW_MULTIPLE_STATEMENTS
static final org.apache.nifi.components.PropertyDescriptor ALLOW_MULTIPLE_STATEMENTS -
QUOTE_IDENTIFIERS
static final org.apache.nifi.components.PropertyDescriptor QUOTE_IDENTIFIERS -
QUOTE_TABLE_IDENTIFIER
static final org.apache.nifi.components.PropertyDescriptor QUOTE_TABLE_IDENTIFIER -
QUERY_TIMEOUT
static final org.apache.nifi.components.PropertyDescriptor QUERY_TIMEOUT -
TABLE_SCHEMA_CACHE_SIZE
static final org.apache.nifi.components.PropertyDescriptor TABLE_SCHEMA_CACHE_SIZE -
MAX_BATCH_SIZE
static final org.apache.nifi.components.PropertyDescriptor MAX_BATCH_SIZE -
AUTO_COMMIT
static final org.apache.nifi.components.PropertyDescriptor AUTO_COMMIT -
DB_TYPE
static final org.apache.nifi.components.PropertyDescriptor DB_TYPE -
DATABASE_DIALECT_SERVICE
static final org.apache.nifi.components.PropertyDescriptor DATABASE_DIALECT_SERVICE -
properties
-
schemaCache
private com.github.benmanes.caffeine.cache.Cache<PutDatabaseRecord.SchemaKey,TableSchema> schemaCache -
databaseDialectService
-
recordPathOperationType
-
dataRecordPath
-
-
Constructor Details
-
PutDatabaseRecord
public PutDatabaseRecord()
-
-
Method Details
-
getRelationships
- Specified by:
getRelationshipsin interfaceorg.apache.nifi.processor.Processor- Overrides:
getRelationshipsin classorg.apache.nifi.processor.AbstractSessionFactoryProcessor
-
getSupportedPropertyDescriptors
- Overrides:
getSupportedPropertyDescriptorsin classorg.apache.nifi.components.AbstractConfigurableComponent
-
customValidate
protected Collection<org.apache.nifi.components.ValidationResult> customValidate(org.apache.nifi.components.ValidationContext validationContext) - Overrides:
customValidatein classorg.apache.nifi.components.AbstractConfigurableComponent
-
isMaxBatchSizeHardcodedToZero
private boolean isMaxBatchSizeHardcodedToZero(org.apache.nifi.components.ValidationContext validationContext) -
onScheduled
@OnScheduled public void onScheduled(org.apache.nifi.processor.ProcessContext context) -
onTrigger
public void onTrigger(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session) throws org.apache.nifi.processor.exception.ProcessException - Specified by:
onTriggerin classorg.apache.nifi.processor.AbstractProcessor- Throws:
org.apache.nifi.processor.exception.ProcessException
-
routeOnException
private void routeOnException(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, Connection connection, Exception e, org.apache.nifi.flowfile.FlowFile flowFile) -
rollbackConnection
-
closeConnection
-
executeSQL
private void executeSQL(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, org.apache.nifi.flowfile.FlowFile flowFile, Connection connection, RecordReader recordReader) throws IllegalArgumentException, MalformedRecordException, IOException, SQLException -
executeDML
private void executeDML(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, org.apache.nifi.flowfile.FlowFile flowFile, Connection con, RecordReader recordReader, String explicitStatementType, PutDatabaseRecord.DMLSettings settings) throws IllegalArgumentException, MalformedRecordException, IOException, SQLException -
setParameter
private void setParameter(PreparedStatement ps, int index, Object value, int fieldSqlType, int sqlType) throws IOException - Throws:
IOException
-
getDataRecords
-
getJdbcUrl
-
getStatementType
private String getStatementType(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.flowfile.FlowFile flowFile) -
validateStatementType
-
putToDatabase
private void putToDatabase(org.apache.nifi.processor.ProcessContext context, org.apache.nifi.processor.ProcessSession session, org.apache.nifi.flowfile.FlowFile flowFile, Connection connection) throws Exception - Throws:
Exception
-
generateTableName
String generateTableName(PutDatabaseRecord.DMLSettings settings, String catalog, String schemaName, String tableName, TableSchema tableSchema) -
getNormalizedColumnNames
private Set<String> getNormalizedColumnNames(RecordSchema schema, boolean translateFieldNames, NameNormalizer normalizer) -
generateInsert
PutDatabaseRecord.SqlAndIncludedColumns generateInsert(RecordSchema recordSchema, String tableName, TableSchema tableSchema, PutDatabaseRecord.DMLSettings settings, NameNormalizer normalizer) throws IllegalArgumentException, SQLException - Throws:
IllegalArgumentExceptionSQLException
-
getSqlStatement
private PutDatabaseRecord.SqlAndIncludedColumns getSqlStatement(StatementType statementType, RecordSchema recordSchema, String qualifiedTableName, String updateKeys, TableSchema tableSchema, PutDatabaseRecord.DMLSettings settings, NameNormalizer normalizer) throws MalformedRecordException, SQLDataException, SQLIntegrityConstraintViolationException -
getColumnDefinition
private ColumnDefinition getColumnDefinition(ColumnDescription columnDescription, Set<String> keyColumnNames, String columnName) -
generateUpdate
PutDatabaseRecord.SqlAndIncludedColumns generateUpdate(RecordSchema recordSchema, String tableName, String updateKeys, TableSchema tableSchema, PutDatabaseRecord.DMLSettings settings, NameNormalizer normalizer) throws IllegalArgumentException, MalformedRecordException, SQLException -
generateDelete
PutDatabaseRecord.SqlAndIncludedColumns generateDelete(RecordSchema recordSchema, String tableName, String deleteKeys, TableSchema tableSchema, PutDatabaseRecord.DMLSettings settings, NameNormalizer normalizer) throws IllegalArgumentException, MalformedRecordException, SQLDataException -
checkValuesForRequiredColumns
private void checkValuesForRequiredColumns(RecordSchema recordSchema, TableSchema tableSchema, PutDatabaseRecord.DMLSettings settings, NameNormalizer normalizer) -
getUpdateKeyColumnNames
private Set<String> getUpdateKeyColumnNames(String tableName, String updateKeys, TableSchema tableSchema) throws SQLIntegrityConstraintViolationException -
normalizeKeyColumnNamesAndCheckForValues
private Set<String> normalizeKeyColumnNamesAndCheckForValues(RecordSchema recordSchema, String updateKeys, PutDatabaseRecord.DMLSettings settings, Set<String> updateKeyColumnNames, NameNormalizer normalizer) throws MalformedRecordException - Throws:
MalformedRecordException
-
isSupportsBatchUpdates
-
getParameterCount
-