You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by mattyb149 <gi...@git.apache.org> on 2016/07/14 01:32:37 UTC
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
GitHub user mattyb149 opened a pull request:
https://github.com/apache/nifi/pull/645
NIFI-2157: Add GenerateTableFetch processor
Refactored common code out of QueryDatabaseTable into an Abstract base class, which involved some refactor of QueryDatabaseTable as well as the added GenerateTableFetch processor.
This includes the addition of the DatabaseAdapter interface and its implementations, going forward this is to isolate database-specific code behind an interface for use by database-related processors.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/mattyb149/nifi NIFI-2157_new
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/645.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #645
----
commit 7b8fd3728f21f6a40b91e401d14738ea9a32fc9e
Author: Matt Burgess <ma...@apache.org>
Date: 2016-07-14T01:29:51Z
NIFI-2157: Add GenerateTableFetch processor
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073752
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
+
+
+ public static final PropertyDescriptor PARTITION_SIZE = new PropertyDescriptor.Builder()
+ .name("gen-table-fetch-partition-size")
+ .displayName("Partition Size")
+ .description("The number of result rows to be fetched by each generated SQL statement. The total number of rows in "
+ + "the table divided by the partition size gives the number of SQL statements (i.e. FlowFiles) generated. A "
+ + "value of zero indicates that a single FlowFile is to be generated whose SQL statement will fetch all rows "
+ + "in the table.")
+ .defaultValue("10000")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .build();
+
+ public GenerateTableFetch() {
+ final Set<Relationship> r = new HashSet<>();
+ r.add(REL_SUCCESS);
+ relationships = Collections.unmodifiableSet(r);
+
+ final List<PropertyDescriptor> pds = new ArrayList<>();
+ pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
+ pds.add(TABLE_NAME);
+ pds.add(COLUMN_NAMES);
+ pds.add(MAX_VALUE_COLUMN_NAMES);
+ pds.add(QUERY_TIMEOUT);
+ pds.add(PARTITION_SIZE);
+ propDescriptors = Collections.unmodifiableList(pds);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propDescriptors;
+ }
+
+ @OnScheduled
+ public void setup(final ProcessContext context) {
+ super.setup(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
+ ProcessSession session = sessionFactory.createSession();
+ final ComponentLog logger = getLogger();
+
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String columnNames = context.getProperty(COLUMN_NAMES).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final int partitionSize = context.getProperty(PARTITION_SIZE).asInteger();
+
+ final StateManager stateManager = context.getStateManager();
+ final StateMap stateMap;
+
+ try {
+ stateMap = stateManager.getState(Scope.CLUSTER);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve observed maximum values from the State Manager. Will not perform "
+ + "query until this is accomplished.", ioe);
+ context.yield();
+ return;
+ }
+ // Make a mutable copy of the current state property map. This will be updated by the result row callback, and eventually
+ // set as the current state map (after the session has been committed)
+ final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
+
+ // Build a WHERE clause with maximum-value columns (if they exist), and a list of column names that will contain MAX(<column>) aliases. The
+ // executed SQL query will retrieve the count of all records after the filter(s) have been applied, as well as the new maximum values for the
+ // specified columns. This allows the processor to generate the correctly partitioned SQL statements as well as to update the state with the
+ // latest observed maximum values.
+ String whereClause = null;
+ List<String> maxValueColumnNameList = StringUtils.isEmpty(maxValueColumnNames)
+ ? new ArrayList<>(0)
+ : Arrays.asList(maxValueColumnNames.split("\\s*,\\s*"));
+ List<String> maxValueClauses = new ArrayList<>(maxValueColumnNameList.size());
+
+ String columnsClause = null;
+ List<String> maxValueSelectColumns = new ArrayList<>(maxValueColumnNameList.size() + 1);
+ maxValueSelectColumns.add("COUNT(*)");
+
+ // For each maximum-value column, get a WHERE filter and a MAX(column) alias
+ maxValueColumnNameList.forEach(colName -> {
+ maxValueSelectColumns.add("MAX(" + colName + ") " + colName);
+ String maxValue = statePropertyMap.get(colName.toLowerCase());
+ if (!StringUtils.isEmpty(maxValue)) {
+ Integer type = columnTypeMap.get(colName.toLowerCase());
+ if (type == null) {
+ // This shouldn't happen as we are populating columnTypeMap when the processor is scheduled.
+ throw new IllegalArgumentException("No column type found for: " + colName);
+ }
+ // Add a condition for the WHERE clause
+ maxValueClauses.add(colName + " > " + getLiteralByType(type, maxValue, dbAdapter.getName()));
+ }
+ });
+
+ whereClause = StringUtils.join(maxValueClauses, " AND ");
+ columnsClause = StringUtils.join(maxValueSelectColumns, ", ");
+
+ // Build a SELECT query with maximum-value columns (if present)
+ final String selectQuery = dbAdapter.getSelectStatement(tableName, columnsClause, whereClause, null, null, null);
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).asTimePeriod(TimeUnit.SECONDS).intValue();
+ st.setQueryTimeout(queryTimeout); // timeout in seconds
+
+ logger.debug("Executing {}", new Object[]{selectQuery});
+ ResultSet resultSet = st.executeQuery(selectQuery);
+ if (resultSet.next()) {
+ // Total row count is in the first column
+ int rowCount = resultSet.getInt(1);
+
+ // Update the state map with the newly-observed maximum values
+ ResultSetMetaData rsmd = resultSet.getMetaData();
+ for (int i = 2; i <= rsmd.getColumnCount(); i++) {
+ String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ int type = rsmd.getColumnType(i);
+ try {
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ if (newMaxValue != null) {
+ statePropertyMap.put(resultColumnName, newMaxValue);
+ }
+ } catch (ParseException | IOException pie) {
+ // Fail the whole thing here before we start creating flow files and such
+ throw new ProcessException(pie);
+ }
+ }
+
+ final int numberOfFetches = (partitionSize == 0) ? rowCount : (rowCount / partitionSize) + (rowCount % partitionSize == 0 ? 0 : 1);
+
+ // Generate SQL statements to read "pages" of data
+ for (int i = 0; i < numberOfFetches; i++) {
+ FlowFile sqlFlowFile = null;
+ try {
+ Integer limit = partitionSize == 0 ? null : partitionSize;
+ Integer offset = partitionSize == 0 ? null : i * partitionSize;
+ final String query = dbAdapter.getSelectStatement(tableName, columnNames, StringUtils.join(maxValueClauses, " AND "), null, limit, offset);
+ sqlFlowFile = session.create();
+ sqlFlowFile = session.write(sqlFlowFile, out -> {
+ out.write(query.getBytes());
+ });
+ session.transfer(sqlFlowFile, REL_SUCCESS);
+
+ } catch (Exception e) {
+ logger.error("Error while generating SQL statement", e);
+ if (sqlFlowFile != null) {
+ session.remove(sqlFlowFile);
+ }
+ }
+ }
+
+ } else {
+ // Something is very wrong here, one row (even if count is zero) should be returned
+ throw new SQLException("No rows returned from metadata query: " + selectQuery);
+ }
+
+
+ session.commit();
+ try {
+ // Update the state
+ stateManager.setState(statePropertyMap, Scope.CLUSTER);
+ } catch (IOException ioe) {
+ getLogger().error("{} failed to update State Manager, observed maximum values will not be recorded", new Object[]{this, ioe});
--- End diff --
This should re-enforce to the user that sql statement will be duplicated.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073659
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
--- End diff --
Yes both good points, and the WritesAttribute should be removed too
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073732
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
+
+
+ public static final PropertyDescriptor PARTITION_SIZE = new PropertyDescriptor.Builder()
+ .name("gen-table-fetch-partition-size")
+ .displayName("Partition Size")
+ .description("The number of result rows to be fetched by each generated SQL statement. The total number of rows in "
+ + "the table divided by the partition size gives the number of SQL statements (i.e. FlowFiles) generated. A "
+ + "value of zero indicates that a single FlowFile is to be generated whose SQL statement will fetch all rows "
+ + "in the table.")
+ .defaultValue("10000")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .build();
+
+ public GenerateTableFetch() {
+ final Set<Relationship> r = new HashSet<>();
+ r.add(REL_SUCCESS);
+ relationships = Collections.unmodifiableSet(r);
+
+ final List<PropertyDescriptor> pds = new ArrayList<>();
+ pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
+ pds.add(TABLE_NAME);
+ pds.add(COLUMN_NAMES);
+ pds.add(MAX_VALUE_COLUMN_NAMES);
+ pds.add(QUERY_TIMEOUT);
+ pds.add(PARTITION_SIZE);
+ propDescriptors = Collections.unmodifiableList(pds);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propDescriptors;
+ }
+
+ @OnScheduled
+ public void setup(final ProcessContext context) {
+ super.setup(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
+ ProcessSession session = sessionFactory.createSession();
+ final ComponentLog logger = getLogger();
+
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String columnNames = context.getProperty(COLUMN_NAMES).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final int partitionSize = context.getProperty(PARTITION_SIZE).asInteger();
+
+ final StateManager stateManager = context.getStateManager();
+ final StateMap stateMap;
+
+ try {
+ stateMap = stateManager.getState(Scope.CLUSTER);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve observed maximum values from the State Manager. Will not perform "
+ + "query until this is accomplished.", ioe);
+ context.yield();
+ return;
+ }
+ // Make a mutable copy of the current state property map. This will be updated by the result row callback, and eventually
+ // set as the current state map (after the session has been committed)
+ final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
+
+ // Build a WHERE clause with maximum-value columns (if they exist), and a list of column names that will contain MAX(<column>) aliases. The
+ // executed SQL query will retrieve the count of all records after the filter(s) have been applied, as well as the new maximum values for the
+ // specified columns. This allows the processor to generate the correctly partitioned SQL statements as well as to update the state with the
+ // latest observed maximum values.
+ String whereClause = null;
+ List<String> maxValueColumnNameList = StringUtils.isEmpty(maxValueColumnNames)
+ ? new ArrayList<>(0)
+ : Arrays.asList(maxValueColumnNames.split("\\s*,\\s*"));
+ List<String> maxValueClauses = new ArrayList<>(maxValueColumnNameList.size());
+
+ String columnsClause = null;
+ List<String> maxValueSelectColumns = new ArrayList<>(maxValueColumnNameList.size() + 1);
+ maxValueSelectColumns.add("COUNT(*)");
+
+ // For each maximum-value column, get a WHERE filter and a MAX(column) alias
+ maxValueColumnNameList.forEach(colName -> {
+ maxValueSelectColumns.add("MAX(" + colName + ") " + colName);
+ String maxValue = statePropertyMap.get(colName.toLowerCase());
+ if (!StringUtils.isEmpty(maxValue)) {
+ Integer type = columnTypeMap.get(colName.toLowerCase());
+ if (type == null) {
+ // This shouldn't happen as we are populating columnTypeMap when the processor is scheduled.
+ throw new IllegalArgumentException("No column type found for: " + colName);
+ }
+ // Add a condition for the WHERE clause
+ maxValueClauses.add(colName + " > " + getLiteralByType(type, maxValue, dbAdapter.getName()));
+ }
+ });
+
+ whereClause = StringUtils.join(maxValueClauses, " AND ");
+ columnsClause = StringUtils.join(maxValueSelectColumns, ", ");
+
+ // Build a SELECT query with maximum-value columns (if present)
+ final String selectQuery = dbAdapter.getSelectStatement(tableName, columnsClause, whereClause, null, null, null);
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).asTimePeriod(TimeUnit.SECONDS).intValue();
+ st.setQueryTimeout(queryTimeout); // timeout in seconds
+
+ logger.debug("Executing {}", new Object[]{selectQuery});
+ ResultSet resultSet = st.executeQuery(selectQuery);
+ if (resultSet.next()) {
+ // Total row count is in the first column
+ int rowCount = resultSet.getInt(1);
+
+ // Update the state map with the newly-observed maximum values
+ ResultSetMetaData rsmd = resultSet.getMetaData();
+ for (int i = 2; i <= rsmd.getColumnCount(); i++) {
+ String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ int type = rsmd.getColumnType(i);
+ try {
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ if (newMaxValue != null) {
+ statePropertyMap.put(resultColumnName, newMaxValue);
--- End diff --
Shouldn't the state map be update after sending the flowfiles? In it's current flow, a failure caught by line 227 would result in sql queries for pages of data being lost.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71087671
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,253 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.TriggerSerially;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@TriggerSerially
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
--- End diff --
Probably. I wasn't sure whether to be liberal or conservative with the SeeAlso annotation
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71074344
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -184,23 +90,18 @@
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
.build();
-
- private final List<PropertyDescriptor> propDescriptors;
-
- protected final Map<String, Integer> columnTypeMap = new HashMap<>();
-
public QueryDatabaseTable() {
final Set<Relationship> r = new HashSet<>();
r.add(REL_SUCCESS);
relationships = Collections.unmodifiableSet(r);
final List<PropertyDescriptor> pds = new ArrayList<>();
pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
pds.add(TABLE_NAME);
pds.add(COLUMN_NAMES);
pds.add(MAX_VALUE_COLUMN_NAMES);
pds.add(QUERY_TIMEOUT);
- pds.add(SQL_PREPROCESS_STRATEGY);
--- End diff --
Ah, it's because it was extended in the DB_TYPE property.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71087619
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,253 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.TriggerSerially;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@TriggerSerially
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums. This "
+ + "processor is intended to be run on the Primary Node only.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
+
+ public static final PropertyDescriptor PARTITION_SIZE = new PropertyDescriptor.Builder()
+ .name("gen-table-fetch-partition-size")
+ .displayName("Partition Size")
+ .description("The number of result rows to be fetched by each generated SQL statement. The total number of rows in "
+ + "the table divided by the partition size gives the number of SQL statements (i.e. FlowFiles) generated. A "
+ + "value of zero indicates that a single FlowFile is to be generated whose SQL statement will fetch all rows "
+ + "in the table.")
+ .defaultValue("10000")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .build();
+
+ public GenerateTableFetch() {
+ final Set<Relationship> r = new HashSet<>();
+ r.add(REL_SUCCESS);
+ relationships = Collections.unmodifiableSet(r);
+
+ final List<PropertyDescriptor> pds = new ArrayList<>();
+ pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
+ pds.add(TABLE_NAME);
+ pds.add(COLUMN_NAMES);
+ pds.add(MAX_VALUE_COLUMN_NAMES);
+ pds.add(QUERY_TIMEOUT);
+ pds.add(PARTITION_SIZE);
+ propDescriptors = Collections.unmodifiableList(pds);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propDescriptors;
+ }
+
+ @OnScheduled
+ public void setup(final ProcessContext context) {
+ super.setup(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
+ ProcessSession session = sessionFactory.createSession();
+ final ComponentLog logger = getLogger();
+
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String columnNames = context.getProperty(COLUMN_NAMES).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final int partitionSize = context.getProperty(PARTITION_SIZE).asInteger();
+
+ final StateManager stateManager = context.getStateManager();
+ final StateMap stateMap;
+
+ try {
+ stateMap = stateManager.getState(Scope.CLUSTER);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve observed maximum values from the State Manager. Will not perform "
+ + "query until this is accomplished.", ioe);
+ context.yield();
+ return;
+ }
+ try {
+ // Make a mutable copy of the current state property map. This will be updated by the result row callback, and eventually
+ // set as the current state map (after the session has been committed)
+ final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
+
+ // Build a WHERE clause with maximum-value columns (if they exist), and a list of column names that will contain MAX(<column>) aliases. The
+ // executed SQL query will retrieve the count of all records after the filter(s) have been applied, as well as the new maximum values for the
+ // specified columns. This allows the processor to generate the correctly partitioned SQL statements as well as to update the state with the
+ // latest observed maximum values.
+ String whereClause = null;
+ List<String> maxValueColumnNameList = StringUtils.isEmpty(maxValueColumnNames)
+ ? new ArrayList<>(0)
+ : Arrays.asList(maxValueColumnNames.split("\\s*,\\s*"));
+ List<String> maxValueClauses = new ArrayList<>(maxValueColumnNameList.size());
+
+ String columnsClause = null;
+ List<String> maxValueSelectColumns = new ArrayList<>(maxValueColumnNameList.size() + 1);
+ maxValueSelectColumns.add("COUNT(*)");
+
+ // For each maximum-value column, get a WHERE filter and a MAX(column) alias
+ maxValueColumnNameList.forEach(colName -> {
+ maxValueSelectColumns.add("MAX(" + colName + ") " + colName);
+ String maxValue = statePropertyMap.get(colName.toLowerCase());
+ if (!StringUtils.isEmpty(maxValue)) {
+ Integer type = columnTypeMap.get(colName.toLowerCase());
+ if (type == null) {
+ // This shouldn't happen as we are populating columnTypeMap when the processor is scheduled.
+ throw new IllegalArgumentException("No column type found for: " + colName);
+ }
+ // Add a condition for the WHERE clause
+ maxValueClauses.add(colName + " > " + getLiteralByType(type, maxValue, dbAdapter.getName()));
+ }
+ });
+
+ whereClause = StringUtils.join(maxValueClauses, " AND ");
+ columnsClause = StringUtils.join(maxValueSelectColumns, ", ");
+
+ // Build a SELECT query with maximum-value columns (if present)
+ final String selectQuery = dbAdapter.getSelectStatement(tableName, columnsClause, whereClause, null, null, null);
+ int rowCount = 0;
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).asTimePeriod(TimeUnit.SECONDS).intValue();
+ st.setQueryTimeout(queryTimeout); // timeout in seconds
+
+ logger.debug("Executing {}", new Object[]{selectQuery});
+ ResultSet resultSet;
+
+ resultSet = st.executeQuery(selectQuery);
+
+ if (resultSet.next()) {
+ // Total row count is in the first column
+ rowCount = resultSet.getInt(1);
+
+ // Update the state map with the newly-observed maximum values
+ ResultSetMetaData rsmd = resultSet.getMetaData();
+ for (int i = 2; i <= rsmd.getColumnCount(); i++) {
+ String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ int type = rsmd.getColumnType(i);
+ try {
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ if (newMaxValue != null) {
+ statePropertyMap.put(resultColumnName, newMaxValue);
+ }
+ } catch (ParseException | IOException pie) {
+ // Fail the whole thing here before we start creating flow files and such
+ throw new ProcessException(pie);
+ }
+ }
+ } else {
+ // Something is very wrong here, one row (even if count is zero) should be returned
+ throw new SQLException("No rows returned from metadata query: " + selectQuery);
+ }
+ } catch (SQLException e) {
+ logger.error("Unable to execute SQL select query {} due to {}", new Object[]{selectQuery, e});
+ throw new ProcessException(e);
+ }
+ final int numberOfFetches = (partitionSize == 0) ? rowCount : (rowCount / partitionSize) + (rowCount % partitionSize == 0 ? 0 : 1);
+
+
+ // Generate SQL statements to read "pages" of data
+ for (int i = 0; i < numberOfFetches; i++) {
+ FlowFile sqlFlowFile;
+
+ Integer limit = partitionSize == 0 ? null : partitionSize;
+ Integer offset = partitionSize == 0 ? null : i * partitionSize;
+ final String query = dbAdapter.getSelectStatement(tableName, columnNames, StringUtils.join(maxValueClauses, " AND "), null, limit, offset);
+ sqlFlowFile = session.create();
+ sqlFlowFile = session.write(sqlFlowFile, out -> {
+ out.write(query.getBytes());
+ });
+ session.transfer(sqlFlowFile, REL_SUCCESS);
+ }
+
+ session.commit();
+ try {
+ // Update the state
+ stateManager.setState(statePropertyMap, Scope.CLUSTER);
+ } catch (IOException ioe) {
+ getLogger().error("{} failed to update State Manager, observed maximum values will not be recorded. "
+ + "Also, any generated SQL statements may be duplicated.",
+ new Object[]{this, ioe});
+ }
+ } catch (final ProcessException pe) {
+ session.rollback();
--- End diff --
This should log something related to the exception it catches
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71074319
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -184,23 +90,18 @@
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
.build();
-
- private final List<PropertyDescriptor> propDescriptors;
-
- protected final Map<String, Integer> columnTypeMap = new HashMap<>();
-
public QueryDatabaseTable() {
final Set<Relationship> r = new HashSet<>();
r.add(REL_SUCCESS);
relationships = Collections.unmodifiableSet(r);
final List<PropertyDescriptor> pds = new ArrayList<>();
pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
pds.add(TABLE_NAME);
pds.add(COLUMN_NAMES);
pds.add(MAX_VALUE_COLUMN_NAMES);
pds.add(QUERY_TIMEOUT);
- pds.add(SQL_PREPROCESS_STRATEGY);
--- End diff --
Why is the SQL_PREPROCESS_STRATEGY being removed from the QueryDatabaseTable processor?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71074368
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -184,23 +90,18 @@
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
.build();
-
- private final List<PropertyDescriptor> propDescriptors;
-
- protected final Map<String, Integer> columnTypeMap = new HashMap<>();
-
public QueryDatabaseTable() {
final Set<Relationship> r = new HashSet<>();
r.add(REL_SUCCESS);
relationships = Collections.unmodifiableSet(r);
final List<PropertyDescriptor> pds = new ArrayList<>();
pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
pds.add(TABLE_NAME);
pds.add(COLUMN_NAMES);
pds.add(MAX_VALUE_COLUMN_NAMES);
pds.add(QUERY_TIMEOUT);
- pds.add(SQL_PREPROCESS_STRATEGY);
--- End diff --
Good point, agreed
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073757
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
+
+
+ public static final PropertyDescriptor PARTITION_SIZE = new PropertyDescriptor.Builder()
+ .name("gen-table-fetch-partition-size")
+ .displayName("Partition Size")
+ .description("The number of result rows to be fetched by each generated SQL statement. The total number of rows in "
+ + "the table divided by the partition size gives the number of SQL statements (i.e. FlowFiles) generated. A "
+ + "value of zero indicates that a single FlowFile is to be generated whose SQL statement will fetch all rows "
+ + "in the table.")
+ .defaultValue("10000")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .build();
+
+ public GenerateTableFetch() {
+ final Set<Relationship> r = new HashSet<>();
+ r.add(REL_SUCCESS);
+ relationships = Collections.unmodifiableSet(r);
+
+ final List<PropertyDescriptor> pds = new ArrayList<>();
+ pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
+ pds.add(TABLE_NAME);
+ pds.add(COLUMN_NAMES);
+ pds.add(MAX_VALUE_COLUMN_NAMES);
+ pds.add(QUERY_TIMEOUT);
+ pds.add(PARTITION_SIZE);
+ propDescriptors = Collections.unmodifiableList(pds);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propDescriptors;
+ }
+
+ @OnScheduled
+ public void setup(final ProcessContext context) {
+ super.setup(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
+ ProcessSession session = sessionFactory.createSession();
+ final ComponentLog logger = getLogger();
+
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String columnNames = context.getProperty(COLUMN_NAMES).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final int partitionSize = context.getProperty(PARTITION_SIZE).asInteger();
+
+ final StateManager stateManager = context.getStateManager();
+ final StateMap stateMap;
+
+ try {
+ stateMap = stateManager.getState(Scope.CLUSTER);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve observed maximum values from the State Manager. Will not perform "
+ + "query until this is accomplished.", ioe);
+ context.yield();
+ return;
+ }
+ // Make a mutable copy of the current state property map. This will be updated by the result row callback, and eventually
+ // set as the current state map (after the session has been committed)
+ final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
+
+ // Build a WHERE clause with maximum-value columns (if they exist), and a list of column names that will contain MAX(<column>) aliases. The
+ // executed SQL query will retrieve the count of all records after the filter(s) have been applied, as well as the new maximum values for the
+ // specified columns. This allows the processor to generate the correctly partitioned SQL statements as well as to update the state with the
+ // latest observed maximum values.
+ String whereClause = null;
+ List<String> maxValueColumnNameList = StringUtils.isEmpty(maxValueColumnNames)
+ ? new ArrayList<>(0)
+ : Arrays.asList(maxValueColumnNames.split("\\s*,\\s*"));
+ List<String> maxValueClauses = new ArrayList<>(maxValueColumnNameList.size());
+
+ String columnsClause = null;
+ List<String> maxValueSelectColumns = new ArrayList<>(maxValueColumnNameList.size() + 1);
+ maxValueSelectColumns.add("COUNT(*)");
+
+ // For each maximum-value column, get a WHERE filter and a MAX(column) alias
+ maxValueColumnNameList.forEach(colName -> {
+ maxValueSelectColumns.add("MAX(" + colName + ") " + colName);
+ String maxValue = statePropertyMap.get(colName.toLowerCase());
+ if (!StringUtils.isEmpty(maxValue)) {
+ Integer type = columnTypeMap.get(colName.toLowerCase());
+ if (type == null) {
+ // This shouldn't happen as we are populating columnTypeMap when the processor is scheduled.
+ throw new IllegalArgumentException("No column type found for: " + colName);
+ }
+ // Add a condition for the WHERE clause
+ maxValueClauses.add(colName + " > " + getLiteralByType(type, maxValue, dbAdapter.getName()));
+ }
+ });
+
+ whereClause = StringUtils.join(maxValueClauses, " AND ");
+ columnsClause = StringUtils.join(maxValueSelectColumns, ", ");
+
+ // Build a SELECT query with maximum-value columns (if present)
+ final String selectQuery = dbAdapter.getSelectStatement(tableName, columnsClause, whereClause, null, null, null);
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).asTimePeriod(TimeUnit.SECONDS).intValue();
+ st.setQueryTimeout(queryTimeout); // timeout in seconds
+
+ logger.debug("Executing {}", new Object[]{selectQuery});
+ ResultSet resultSet = st.executeQuery(selectQuery);
+ if (resultSet.next()) {
+ // Total row count is in the first column
+ int rowCount = resultSet.getInt(1);
+
+ // Update the state map with the newly-observed maximum values
+ ResultSetMetaData rsmd = resultSet.getMetaData();
+ for (int i = 2; i <= rsmd.getColumnCount(); i++) {
+ String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ int type = rsmd.getColumnType(i);
+ try {
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ if (newMaxValue != null) {
+ statePropertyMap.put(resultColumnName, newMaxValue);
+ }
+ } catch (ParseException | IOException pie) {
+ // Fail the whole thing here before we start creating flow files and such
+ throw new ProcessException(pie);
+ }
+ }
+
+ final int numberOfFetches = (partitionSize == 0) ? rowCount : (rowCount / partitionSize) + (rowCount % partitionSize == 0 ? 0 : 1);
+
+ // Generate SQL statements to read "pages" of data
+ for (int i = 0; i < numberOfFetches; i++) {
+ FlowFile sqlFlowFile = null;
+ try {
+ Integer limit = partitionSize == 0 ? null : partitionSize;
+ Integer offset = partitionSize == 0 ? null : i * partitionSize;
+ final String query = dbAdapter.getSelectStatement(tableName, columnNames, StringUtils.join(maxValueClauses, " AND "), null, limit, offset);
+ sqlFlowFile = session.create();
+ sqlFlowFile = session.write(sqlFlowFile, out -> {
+ out.write(query.getBytes());
+ });
+ session.transfer(sqlFlowFile, REL_SUCCESS);
+
+ } catch (Exception e) {
+ logger.error("Error while generating SQL statement", e);
+ if (sqlFlowFile != null) {
+ session.remove(sqlFlowFile);
--- End diff --
Definitely possible, but not sure how one page would fail when the others wouldn't, unless there's a bug in the code (which I should fix). If exceptions could happen for arbitrary pages, I will move the try/catch outside
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/645
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71074419
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java ---
@@ -0,0 +1,361 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.processor.AbstractSessionFactoryProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.text.DecimalFormat;
+import java.text.ParseException;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.ServiceLoader;
+import java.util.Set;
+
+import static java.sql.Types.ARRAY;
+import static java.sql.Types.BIGINT;
+import static java.sql.Types.BINARY;
+import static java.sql.Types.BIT;
+import static java.sql.Types.BLOB;
+import static java.sql.Types.BOOLEAN;
+import static java.sql.Types.CLOB;
+import static java.sql.Types.DECIMAL;
+import static java.sql.Types.DOUBLE;
+import static java.sql.Types.FLOAT;
+import static java.sql.Types.INTEGER;
+import static java.sql.Types.LONGVARBINARY;
+import static java.sql.Types.NUMERIC;
+import static java.sql.Types.CHAR;
+import static java.sql.Types.DATE;
+import static java.sql.Types.LONGNVARCHAR;
+import static java.sql.Types.LONGVARCHAR;
+import static java.sql.Types.NCHAR;
+import static java.sql.Types.NVARCHAR;
+import static java.sql.Types.REAL;
+import static java.sql.Types.ROWID;
+import static java.sql.Types.SMALLINT;
+import static java.sql.Types.TIME;
+import static java.sql.Types.TIMESTAMP;
+import static java.sql.Types.TINYINT;
+import static java.sql.Types.VARBINARY;
+import static java.sql.Types.VARCHAR;
+
+/**
+ * A base class for common code shared by processors that fetch RDBMS data.
+ */
+public abstract class AbstractDatabaseFetchProcessor extends AbstractSessionFactoryProcessor {
+
+ // Relationships
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("Successfully created FlowFile from SQL query result set.")
+ .build();
+
+ protected Set<Relationship> relationships;
+
+ // Properties
+ public static final PropertyDescriptor DBCP_SERVICE = new PropertyDescriptor.Builder()
+ .name("Database Connection Pooling Service")
+ .description("The Controller Service that is used to obtain a connection to the database.")
+ .required(true)
+ .identifiesControllerService(DBCPService.class)
+ .build();
+
+ public static final PropertyDescriptor TABLE_NAME = new PropertyDescriptor.Builder()
+ .name("Table Name")
+ .description("The name of the database table to be queried.")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor COLUMN_NAMES = new PropertyDescriptor.Builder()
+ .name("Columns to Return")
+ .description("A comma-separated list of column names to be used in the query. If your database requires "
+ + "special treatment of the names (quoting, e.g.), each name should include such treatment. If no "
+ + "column names are supplied, all columns in the specified table will be returned.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor MAX_VALUE_COLUMN_NAMES = new PropertyDescriptor.Builder()
+ .name("Maximum-value Columns")
+ .description("A comma-separated list of column names. The processor will keep track of the maximum value "
+ + "for each column that has been returned since the processor started running. This can be used to "
+ + "retrieve only those rows that have been added/updated since the last retrieval. Note that some "
+ + "JDBC types such as bit/boolean are not conducive to maintaining maximum value, so columns of these "
+ + "types should not be listed in this property, and will result in error(s) during processing.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor QUERY_TIMEOUT = new PropertyDescriptor.Builder()
+ .name("Max Wait Time")
+ .description("The maximum amount of time allowed for a running SQL select query "
+ + ", zero means there is no limit. Max time less than 1 second will be equal to zero.")
+ .defaultValue("0 seconds")
+ .required(true)
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .build();
+
+ protected List<PropertyDescriptor> propDescriptors;
+
+ public static final PropertyDescriptor DB_TYPE;
+
+ protected final static Map<String, DatabaseAdapter> dbAdapters = new HashMap<>();
+ protected final Map<String, Integer> columnTypeMap = new HashMap<>();
+
+ static {
+ // Load the DatabaseAdapters
+ ServiceLoader<DatabaseAdapter> dbAdapterLoader = ServiceLoader.load(DatabaseAdapter.class);
+ dbAdapterLoader.forEach(it -> dbAdapters.put(it.getName(), it));
+
+ DB_TYPE = new PropertyDescriptor.Builder()
+ .name("db-fetch-db-type")
+ .displayName("Database Type")
+ .description("The type/flavor of database, used for generating database-specific code. In many cases the Generic type "
+ + "should suffice, but some databases (such as Oracle) require custom SQL clauses. ")
+ .allowableValues(dbAdapters.keySet())
+ .defaultValue(dbAdapters.values().stream().findFirst().get().getName())
+ .required(true)
+ .build();
+ }
+
+ public void setup(final ProcessContext context) {
+ // Try to fill the columnTypeMap with the types of the desired max-value columns
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ // Try a query that returns no rows, for the purposes of getting metadata about the columns. It is possible
+ // to use DatabaseMetaData.getColumns(), but not all drivers support this, notably the schema-on-read
+ // approach as in Apache Drill
+ String query = dbAdapter.getSelectStatement(tableName, maxValueColumnNames, "1 = 0", null, null, null);
+ ResultSet resultSet = st.executeQuery(query);
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ int numCols = resultSetMetaData.getColumnCount();
+ if (numCols > 0) {
+ columnTypeMap.clear();
+ for (int i = 1; i <= numCols; i++) {
+ String colName = resultSetMetaData.getColumnName(i).toLowerCase();
+ int colType = resultSetMetaData.getColumnType(i);
+ columnTypeMap.put(colName, colType);
+ }
+
+ } else {
+ throw new ProcessException("No columns found in table from those specified: " + maxValueColumnNames);
+ }
+
+ } catch (SQLException e) {
+ throw new ProcessException("Unable to communicate with database in order to determine column types", e);
+ }
+ }
+
+ protected static String getMaxValueFromRow(ResultSet resultSet,
+ int columnIndex,
+ Integer type,
+ String maxValueString,
+ String preProcessStrategy)
+ throws ParseException, IOException, SQLException {
+
+ // Skip any columns we're not keeping track of or whose value is null
+ if (type == null || resultSet.getObject(columnIndex) == null) {
+ return null;
+ }
+
+ switch (type) {
+ case CHAR:
+ case LONGNVARCHAR:
+ case LONGVARCHAR:
+ case NCHAR:
+ case NVARCHAR:
+ case VARCHAR:
+ case ROWID:
+ String colStringValue = resultSet.getString(columnIndex);
+ if (maxValueString == null || colStringValue.compareTo(maxValueString) > 0) {
+ return colStringValue;
+ }
+ break;
+
+ case INTEGER:
+ case SMALLINT:
+ case TINYINT:
+ Integer colIntValue = resultSet.getInt(columnIndex);
+ Integer maxIntValue = null;
+ if (maxValueString != null) {
+ maxIntValue = Integer.valueOf(maxValueString);
+ }
+ if (maxIntValue == null || colIntValue > maxIntValue) {
+ return colIntValue.toString();
+ }
+ break;
+
+ case BIGINT:
+ Long colLongValue = resultSet.getLong(columnIndex);
+ Long maxLongValue = null;
+ if (maxValueString != null) {
+ maxLongValue = Long.valueOf(maxValueString);
+ }
+ if (maxLongValue == null || colLongValue > maxLongValue) {
+ return colLongValue.toString();
+ }
+ break;
+
+ case FLOAT:
+ case REAL:
+ case DOUBLE:
+ Double colDoubleValue = resultSet.getDouble(columnIndex);
+ Double maxDoubleValue = null;
+ if (maxValueString != null) {
+ maxDoubleValue = Double.valueOf(maxValueString);
+ }
+ if (maxDoubleValue == null || colDoubleValue > maxDoubleValue) {
+ return colDoubleValue.toString();
+ }
+ break;
+
+ case DECIMAL:
+ case NUMERIC:
+ BigDecimal colBigDecimalValue = resultSet.getBigDecimal(columnIndex);
+ BigDecimal maxBigDecimalValue = null;
+ if (maxValueString != null) {
+ DecimalFormat df = new DecimalFormat();
+ df.setParseBigDecimal(true);
+ maxBigDecimalValue = (BigDecimal) df.parse(maxValueString);
+ }
+ if (maxBigDecimalValue == null || colBigDecimalValue.compareTo(maxBigDecimalValue) > 0) {
+ return colBigDecimalValue.toString();
+ }
+ break;
+
+ case DATE:
+ Date rawColDateValue = resultSet.getDate(columnIndex);
+ java.sql.Date colDateValue = new java.sql.Date(rawColDateValue.getTime());
+ java.sql.Date maxDateValue = null;
+ if (maxValueString != null) {
+ maxDateValue = java.sql.Date.valueOf(maxValueString);
+ }
+ if (maxDateValue == null || colDateValue.after(maxDateValue)) {
+ return colDateValue.toString();
+ }
+ break;
+
+ case TIME:
+ Date rawColTimeValue = resultSet.getDate(columnIndex);
+ java.sql.Time colTimeValue = new java.sql.Time(rawColTimeValue.getTime());
+ java.sql.Time maxTimeValue = null;
+ if (maxValueString != null) {
+ maxTimeValue = java.sql.Time.valueOf(maxValueString);
+ }
+ if (maxTimeValue == null || colTimeValue.after(maxTimeValue)) {
+ return colTimeValue.toString();
+ }
+ break;
+
+ case TIMESTAMP:
+ // Oracle timestamp queries must use literals in java.sql.Date format
+ if ("Oracle".equals(preProcessStrategy)) {
+ Date rawColOracleTimestampValue = resultSet.getDate(columnIndex);
+ java.sql.Date oracleTimestampValue = new java.sql.Date(rawColOracleTimestampValue.getTime());
+ java.sql.Date maxOracleTimestampValue = null;
+ if (maxValueString != null) {
+ maxOracleTimestampValue = java.sql.Date.valueOf(maxValueString);
+ }
+ if (maxOracleTimestampValue == null || oracleTimestampValue.after(maxOracleTimestampValue)) {
+ return oracleTimestampValue.toString();
+ }
+ } else {
+ Timestamp rawColTimestampValue = resultSet.getTimestamp(columnIndex);
+ java.sql.Timestamp colTimestampValue = new java.sql.Timestamp(rawColTimestampValue.getTime());
+ java.sql.Timestamp maxTimestampValue = null;
+ if (maxValueString != null) {
+ maxTimestampValue = java.sql.Timestamp.valueOf(maxValueString);
+ }
+ if (maxTimestampValue == null || colTimestampValue.after(maxTimestampValue)) {
+ return colTimestampValue.toString();
+ }
+ }
+ break;
+
+ case BIT:
+ case BOOLEAN:
+ case BINARY:
+ case VARBINARY:
+ case LONGVARBINARY:
+ case ARRAY:
+ case BLOB:
+ case CLOB:
+ default:
+ throw new IOException("Type for column " + columnIndex + " is not valid for maintaining maximum value");
+ }
+ return null;
+ }
+
+ /**
+ * Returns a SQL literal for the given value based on its type. For example, values of character type need to be enclosed
+ * in single quotes, whereas values of numeric type should not be.
+ *
+ * @param type The JDBC type for the desired literal
+ * @param value The value to be converted to a SQL literal
+ * @return A String representing the given value as a literal of the given type
+ */
+ protected static String getLiteralByType(int type, String value, String preProcessStrategy) {
--- End diff --
The last parameter shouldn't be "preProcessStrategy" since that property was removed in favor of database type.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on the issue:
https://github.com/apache/nifi/pull/645
@mattyb149, just finished doing a simple verification of functionality using MySQL. I set up a flow to grab from randomapi.com changed one of the fields to an incrementing number and inserted it using PutSQL. I then observed GenerateTableFetch create SELECT statements for over those events which I then passed to ExecuteSQL.
Once the comments above are addressed it looks good to me.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71087628
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,253 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.TriggerSerially;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@TriggerSerially
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
--- End diff --
Should the other SQL processors (especially the ones that could accept this as input) be added?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on the issue:
https://github.com/apache/nifi/pull/645
+1
Went through code and any comments were addressed. ran a contrib check build and verified against a MySQL DB. Thanks @mattyb149, I will merge it in.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71074365
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -184,23 +90,18 @@
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
.build();
-
- private final List<PropertyDescriptor> propDescriptors;
-
- protected final Map<String, Integer> columnTypeMap = new HashMap<>();
-
public QueryDatabaseTable() {
final Set<Relationship> r = new HashSet<>();
r.add(REL_SUCCESS);
relationships = Collections.unmodifiableSet(r);
final List<PropertyDescriptor> pds = new ArrayList<>();
pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
pds.add(TABLE_NAME);
pds.add(COLUMN_NAMES);
pds.add(MAX_VALUE_COLUMN_NAMES);
pds.add(QUERY_TIMEOUT);
- pds.add(SQL_PREPROCESS_STRATEGY);
--- End diff --
This should be noted in the 1.0 migration guide
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073773
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
+
+
+ public static final PropertyDescriptor PARTITION_SIZE = new PropertyDescriptor.Builder()
+ .name("gen-table-fetch-partition-size")
+ .displayName("Partition Size")
+ .description("The number of result rows to be fetched by each generated SQL statement. The total number of rows in "
+ + "the table divided by the partition size gives the number of SQL statements (i.e. FlowFiles) generated. A "
+ + "value of zero indicates that a single FlowFile is to be generated whose SQL statement will fetch all rows "
+ + "in the table.")
+ .defaultValue("10000")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .build();
+
+ public GenerateTableFetch() {
+ final Set<Relationship> r = new HashSet<>();
+ r.add(REL_SUCCESS);
+ relationships = Collections.unmodifiableSet(r);
+
+ final List<PropertyDescriptor> pds = new ArrayList<>();
+ pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
+ pds.add(TABLE_NAME);
+ pds.add(COLUMN_NAMES);
+ pds.add(MAX_VALUE_COLUMN_NAMES);
+ pds.add(QUERY_TIMEOUT);
+ pds.add(PARTITION_SIZE);
+ propDescriptors = Collections.unmodifiableList(pds);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propDescriptors;
+ }
+
+ @OnScheduled
+ public void setup(final ProcessContext context) {
+ super.setup(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
+ ProcessSession session = sessionFactory.createSession();
+ final ComponentLog logger = getLogger();
+
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String columnNames = context.getProperty(COLUMN_NAMES).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final int partitionSize = context.getProperty(PARTITION_SIZE).asInteger();
+
+ final StateManager stateManager = context.getStateManager();
+ final StateMap stateMap;
+
+ try {
+ stateMap = stateManager.getState(Scope.CLUSTER);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve observed maximum values from the State Manager. Will not perform "
+ + "query until this is accomplished.", ioe);
+ context.yield();
+ return;
+ }
+ // Make a mutable copy of the current state property map. This will be updated by the result row callback, and eventually
+ // set as the current state map (after the session has been committed)
+ final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
+
+ // Build a WHERE clause with maximum-value columns (if they exist), and a list of column names that will contain MAX(<column>) aliases. The
+ // executed SQL query will retrieve the count of all records after the filter(s) have been applied, as well as the new maximum values for the
+ // specified columns. This allows the processor to generate the correctly partitioned SQL statements as well as to update the state with the
+ // latest observed maximum values.
+ String whereClause = null;
+ List<String> maxValueColumnNameList = StringUtils.isEmpty(maxValueColumnNames)
+ ? new ArrayList<>(0)
+ : Arrays.asList(maxValueColumnNames.split("\\s*,\\s*"));
+ List<String> maxValueClauses = new ArrayList<>(maxValueColumnNameList.size());
+
+ String columnsClause = null;
+ List<String> maxValueSelectColumns = new ArrayList<>(maxValueColumnNameList.size() + 1);
+ maxValueSelectColumns.add("COUNT(*)");
+
+ // For each maximum-value column, get a WHERE filter and a MAX(column) alias
+ maxValueColumnNameList.forEach(colName -> {
+ maxValueSelectColumns.add("MAX(" + colName + ") " + colName);
+ String maxValue = statePropertyMap.get(colName.toLowerCase());
+ if (!StringUtils.isEmpty(maxValue)) {
+ Integer type = columnTypeMap.get(colName.toLowerCase());
+ if (type == null) {
+ // This shouldn't happen as we are populating columnTypeMap when the processor is scheduled.
+ throw new IllegalArgumentException("No column type found for: " + colName);
+ }
+ // Add a condition for the WHERE clause
+ maxValueClauses.add(colName + " > " + getLiteralByType(type, maxValue, dbAdapter.getName()));
+ }
+ });
+
+ whereClause = StringUtils.join(maxValueClauses, " AND ");
+ columnsClause = StringUtils.join(maxValueSelectColumns, ", ");
+
+ // Build a SELECT query with maximum-value columns (if present)
+ final String selectQuery = dbAdapter.getSelectStatement(tableName, columnsClause, whereClause, null, null, null);
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).asTimePeriod(TimeUnit.SECONDS).intValue();
+ st.setQueryTimeout(queryTimeout); // timeout in seconds
+
+ logger.debug("Executing {}", new Object[]{selectQuery});
+ ResultSet resultSet = st.executeQuery(selectQuery);
+ if (resultSet.next()) {
+ // Total row count is in the first column
+ int rowCount = resultSet.getInt(1);
+
+ // Update the state map with the newly-observed maximum values
+ ResultSetMetaData rsmd = resultSet.getMetaData();
+ for (int i = 2; i <= rsmd.getColumnCount(); i++) {
+ String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ int type = rsmd.getColumnType(i);
+ try {
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ if (newMaxValue != null) {
+ statePropertyMap.put(resultColumnName, newMaxValue);
+ }
+ } catch (ParseException | IOException pie) {
+ // Fail the whole thing here before we start creating flow files and such
+ throw new ProcessException(pie);
+ }
+ }
+
+ final int numberOfFetches = (partitionSize == 0) ? rowCount : (rowCount / partitionSize) + (rowCount % partitionSize == 0 ? 0 : 1);
+
+ // Generate SQL statements to read "pages" of data
+ for (int i = 0; i < numberOfFetches; i++) {
+ FlowFile sqlFlowFile = null;
+ try {
+ Integer limit = partitionSize == 0 ? null : partitionSize;
+ Integer offset = partitionSize == 0 ? null : i * partitionSize;
+ final String query = dbAdapter.getSelectStatement(tableName, columnNames, StringUtils.join(maxValueClauses, " AND "), null, limit, offset);
+ sqlFlowFile = session.create();
+ sqlFlowFile = session.write(sqlFlowFile, out -> {
+ out.write(query.getBytes());
+ });
+ session.transfer(sqlFlowFile, REL_SUCCESS);
+
+ } catch (Exception e) {
+ logger.error("Error while generating SQL statement", e);
+ if (sqlFlowFile != null) {
+ session.remove(sqlFlowFile);
+ }
+ }
+ }
+
+ } else {
+ // Something is very wrong here, one row (even if count is zero) should be returned
+ throw new SQLException("No rows returned from metadata query: " + selectQuery);
+ }
+
+
+ session.commit();
+ try {
+ // Update the state
+ stateManager.setState(statePropertyMap, Scope.CLUSTER);
+ } catch (IOException ioe) {
+ getLogger().error("{} failed to update State Manager, observed maximum values will not be recorded", new Object[]{this, ioe});
+ }
+ } catch (final ProcessException | SQLException e) {
--- End diff --
Does this try/catch need to be this large? It could fail after the sql query already ran successfully and fail later for an un-related reason and it would be prepended with "Unable to execute SQL select query {} due to {}"
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073480
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
--- End diff --
Also I believe "EventDriven" should be removed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71087720
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,253 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.TriggerSerially;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@TriggerSerially
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
--- End diff --
That's fair, as long as you have a reason for the choice, I'm fine with either.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71074306
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
+
+
+ public static final PropertyDescriptor PARTITION_SIZE = new PropertyDescriptor.Builder()
+ .name("gen-table-fetch-partition-size")
+ .displayName("Partition Size")
+ .description("The number of result rows to be fetched by each generated SQL statement. The total number of rows in "
+ + "the table divided by the partition size gives the number of SQL statements (i.e. FlowFiles) generated. A "
+ + "value of zero indicates that a single FlowFile is to be generated whose SQL statement will fetch all rows "
+ + "in the table.")
+ .defaultValue("10000")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .build();
+
+ public GenerateTableFetch() {
+ final Set<Relationship> r = new HashSet<>();
+ r.add(REL_SUCCESS);
+ relationships = Collections.unmodifiableSet(r);
+
+ final List<PropertyDescriptor> pds = new ArrayList<>();
+ pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
+ pds.add(TABLE_NAME);
+ pds.add(COLUMN_NAMES);
+ pds.add(MAX_VALUE_COLUMN_NAMES);
+ pds.add(QUERY_TIMEOUT);
+ pds.add(PARTITION_SIZE);
+ propDescriptors = Collections.unmodifiableList(pds);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propDescriptors;
+ }
+
+ @OnScheduled
+ public void setup(final ProcessContext context) {
+ super.setup(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
+ ProcessSession session = sessionFactory.createSession();
+ final ComponentLog logger = getLogger();
+
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String columnNames = context.getProperty(COLUMN_NAMES).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final int partitionSize = context.getProperty(PARTITION_SIZE).asInteger();
+
+ final StateManager stateManager = context.getStateManager();
+ final StateMap stateMap;
+
+ try {
+ stateMap = stateManager.getState(Scope.CLUSTER);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve observed maximum values from the State Manager. Will not perform "
+ + "query until this is accomplished.", ioe);
+ context.yield();
+ return;
+ }
+ // Make a mutable copy of the current state property map. This will be updated by the result row callback, and eventually
+ // set as the current state map (after the session has been committed)
+ final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
+
+ // Build a WHERE clause with maximum-value columns (if they exist), and a list of column names that will contain MAX(<column>) aliases. The
+ // executed SQL query will retrieve the count of all records after the filter(s) have been applied, as well as the new maximum values for the
+ // specified columns. This allows the processor to generate the correctly partitioned SQL statements as well as to update the state with the
+ // latest observed maximum values.
+ String whereClause = null;
+ List<String> maxValueColumnNameList = StringUtils.isEmpty(maxValueColumnNames)
+ ? new ArrayList<>(0)
+ : Arrays.asList(maxValueColumnNames.split("\\s*,\\s*"));
+ List<String> maxValueClauses = new ArrayList<>(maxValueColumnNameList.size());
+
+ String columnsClause = null;
+ List<String> maxValueSelectColumns = new ArrayList<>(maxValueColumnNameList.size() + 1);
+ maxValueSelectColumns.add("COUNT(*)");
+
+ // For each maximum-value column, get a WHERE filter and a MAX(column) alias
+ maxValueColumnNameList.forEach(colName -> {
+ maxValueSelectColumns.add("MAX(" + colName + ") " + colName);
+ String maxValue = statePropertyMap.get(colName.toLowerCase());
+ if (!StringUtils.isEmpty(maxValue)) {
+ Integer type = columnTypeMap.get(colName.toLowerCase());
+ if (type == null) {
+ // This shouldn't happen as we are populating columnTypeMap when the processor is scheduled.
+ throw new IllegalArgumentException("No column type found for: " + colName);
+ }
+ // Add a condition for the WHERE clause
+ maxValueClauses.add(colName + " > " + getLiteralByType(type, maxValue, dbAdapter.getName()));
+ }
+ });
+
+ whereClause = StringUtils.join(maxValueClauses, " AND ");
+ columnsClause = StringUtils.join(maxValueSelectColumns, ", ");
+
+ // Build a SELECT query with maximum-value columns (if present)
+ final String selectQuery = dbAdapter.getSelectStatement(tableName, columnsClause, whereClause, null, null, null);
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).asTimePeriod(TimeUnit.SECONDS).intValue();
+ st.setQueryTimeout(queryTimeout); // timeout in seconds
+
+ logger.debug("Executing {}", new Object[]{selectQuery});
+ ResultSet resultSet = st.executeQuery(selectQuery);
+ if (resultSet.next()) {
+ // Total row count is in the first column
+ int rowCount = resultSet.getInt(1);
+
+ // Update the state map with the newly-observed maximum values
+ ResultSetMetaData rsmd = resultSet.getMetaData();
+ for (int i = 2; i <= rsmd.getColumnCount(); i++) {
+ String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ int type = rsmd.getColumnType(i);
+ try {
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ if (newMaxValue != null) {
+ statePropertyMap.put(resultColumnName, newMaxValue);
+ }
+ } catch (ParseException | IOException pie) {
+ // Fail the whole thing here before we start creating flow files and such
+ throw new ProcessException(pie);
+ }
+ }
+
+ final int numberOfFetches = (partitionSize == 0) ? rowCount : (rowCount / partitionSize) + (rowCount % partitionSize == 0 ? 0 : 1);
+
+ // Generate SQL statements to read "pages" of data
+ for (int i = 0; i < numberOfFetches; i++) {
+ FlowFile sqlFlowFile = null;
+ try {
+ Integer limit = partitionSize == 0 ? null : partitionSize;
+ Integer offset = partitionSize == 0 ? null : i * partitionSize;
+ final String query = dbAdapter.getSelectStatement(tableName, columnNames, StringUtils.join(maxValueClauses, " AND "), null, limit, offset);
+ sqlFlowFile = session.create();
+ sqlFlowFile = session.write(sqlFlowFile, out -> {
+ out.write(query.getBytes());
+ });
+ session.transfer(sqlFlowFile, REL_SUCCESS);
+
+ } catch (Exception e) {
+ logger.error("Error while generating SQL statement", e);
+ if (sqlFlowFile != null) {
+ session.remove(sqlFlowFile);
--- End diff --
The session may fail to write the contents of one flowfile but succeed for another for one reason or another. If any fail, I'd suggest wrapping the for loop with the try/catch so it will stop processing and roll back the whole session so no data is lost.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073740
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
+
+
+ public static final PropertyDescriptor PARTITION_SIZE = new PropertyDescriptor.Builder()
+ .name("gen-table-fetch-partition-size")
+ .displayName("Partition Size")
+ .description("The number of result rows to be fetched by each generated SQL statement. The total number of rows in "
+ + "the table divided by the partition size gives the number of SQL statements (i.e. FlowFiles) generated. A "
+ + "value of zero indicates that a single FlowFile is to be generated whose SQL statement will fetch all rows "
+ + "in the table.")
+ .defaultValue("10000")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .build();
+
+ public GenerateTableFetch() {
+ final Set<Relationship> r = new HashSet<>();
+ r.add(REL_SUCCESS);
+ relationships = Collections.unmodifiableSet(r);
+
+ final List<PropertyDescriptor> pds = new ArrayList<>();
+ pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
+ pds.add(TABLE_NAME);
+ pds.add(COLUMN_NAMES);
+ pds.add(MAX_VALUE_COLUMN_NAMES);
+ pds.add(QUERY_TIMEOUT);
+ pds.add(PARTITION_SIZE);
+ propDescriptors = Collections.unmodifiableList(pds);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propDescriptors;
+ }
+
+ @OnScheduled
+ public void setup(final ProcessContext context) {
+ super.setup(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
+ ProcessSession session = sessionFactory.createSession();
+ final ComponentLog logger = getLogger();
+
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String columnNames = context.getProperty(COLUMN_NAMES).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final int partitionSize = context.getProperty(PARTITION_SIZE).asInteger();
+
+ final StateManager stateManager = context.getStateManager();
+ final StateMap stateMap;
+
+ try {
+ stateMap = stateManager.getState(Scope.CLUSTER);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve observed maximum values from the State Manager. Will not perform "
+ + "query until this is accomplished.", ioe);
+ context.yield();
+ return;
+ }
+ // Make a mutable copy of the current state property map. This will be updated by the result row callback, and eventually
+ // set as the current state map (after the session has been committed)
+ final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
+
+ // Build a WHERE clause with maximum-value columns (if they exist), and a list of column names that will contain MAX(<column>) aliases. The
+ // executed SQL query will retrieve the count of all records after the filter(s) have been applied, as well as the new maximum values for the
+ // specified columns. This allows the processor to generate the correctly partitioned SQL statements as well as to update the state with the
+ // latest observed maximum values.
+ String whereClause = null;
+ List<String> maxValueColumnNameList = StringUtils.isEmpty(maxValueColumnNames)
+ ? new ArrayList<>(0)
+ : Arrays.asList(maxValueColumnNames.split("\\s*,\\s*"));
+ List<String> maxValueClauses = new ArrayList<>(maxValueColumnNameList.size());
+
+ String columnsClause = null;
+ List<String> maxValueSelectColumns = new ArrayList<>(maxValueColumnNameList.size() + 1);
+ maxValueSelectColumns.add("COUNT(*)");
+
+ // For each maximum-value column, get a WHERE filter and a MAX(column) alias
+ maxValueColumnNameList.forEach(colName -> {
+ maxValueSelectColumns.add("MAX(" + colName + ") " + colName);
+ String maxValue = statePropertyMap.get(colName.toLowerCase());
+ if (!StringUtils.isEmpty(maxValue)) {
+ Integer type = columnTypeMap.get(colName.toLowerCase());
+ if (type == null) {
+ // This shouldn't happen as we are populating columnTypeMap when the processor is scheduled.
+ throw new IllegalArgumentException("No column type found for: " + colName);
+ }
+ // Add a condition for the WHERE clause
+ maxValueClauses.add(colName + " > " + getLiteralByType(type, maxValue, dbAdapter.getName()));
+ }
+ });
+
+ whereClause = StringUtils.join(maxValueClauses, " AND ");
+ columnsClause = StringUtils.join(maxValueSelectColumns, ", ");
+
+ // Build a SELECT query with maximum-value columns (if present)
+ final String selectQuery = dbAdapter.getSelectStatement(tableName, columnsClause, whereClause, null, null, null);
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).asTimePeriod(TimeUnit.SECONDS).intValue();
+ st.setQueryTimeout(queryTimeout); // timeout in seconds
+
+ logger.debug("Executing {}", new Object[]{selectQuery});
+ ResultSet resultSet = st.executeQuery(selectQuery);
+ if (resultSet.next()) {
+ // Total row count is in the first column
+ int rowCount = resultSet.getInt(1);
+
+ // Update the state map with the newly-observed maximum values
+ ResultSetMetaData rsmd = resultSet.getMetaData();
+ for (int i = 2; i <= rsmd.getColumnCount(); i++) {
+ String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ int type = rsmd.getColumnType(i);
+ try {
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ if (newMaxValue != null) {
+ statePropertyMap.put(resultColumnName, newMaxValue);
+ }
+ } catch (ParseException | IOException pie) {
+ // Fail the whole thing here before we start creating flow files and such
+ throw new ProcessException(pie);
+ }
+ }
+
+ final int numberOfFetches = (partitionSize == 0) ? rowCount : (rowCount / partitionSize) + (rowCount % partitionSize == 0 ? 0 : 1);
+
+ // Generate SQL statements to read "pages" of data
+ for (int i = 0; i < numberOfFetches; i++) {
+ FlowFile sqlFlowFile = null;
+ try {
+ Integer limit = partitionSize == 0 ? null : partitionSize;
+ Integer offset = partitionSize == 0 ? null : i * partitionSize;
+ final String query = dbAdapter.getSelectStatement(tableName, columnNames, StringUtils.join(maxValueClauses, " AND "), null, limit, offset);
+ sqlFlowFile = session.create();
+ sqlFlowFile = session.write(sqlFlowFile, out -> {
+ out.write(query.getBytes());
+ });
+ session.transfer(sqlFlowFile, REL_SUCCESS);
+
+ } catch (Exception e) {
+ logger.error("Error while generating SQL statement", e);
+ if (sqlFlowFile != null) {
+ session.remove(sqlFlowFile);
--- End diff --
By doing the try/catch just over the pages, wouldn't a failure caught here result in queries being lost?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi issue #645: NIFI-2157: Add GenerateTableFetch processor
Posted by simonellistonball <gi...@git.apache.org>.
Github user simonellistonball commented on the issue:
https://github.com/apache/nifi/pull/645
The description of partition size here, and indeed the name seems a little unclear. Wouldn't partitionCount make more sense, since that is what it actually represents? As stands, it is not the size, or the number of rows, it is the devisor in the function that gives us rough number of rows.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073650
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java ---
@@ -0,0 +1,361 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.processor.AbstractSessionFactoryProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.text.DecimalFormat;
+import java.text.ParseException;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.ServiceLoader;
+import java.util.Set;
+
+import static java.sql.Types.ARRAY;
+import static java.sql.Types.BIGINT;
+import static java.sql.Types.BINARY;
+import static java.sql.Types.BIT;
+import static java.sql.Types.BLOB;
+import static java.sql.Types.BOOLEAN;
+import static java.sql.Types.CLOB;
+import static java.sql.Types.DECIMAL;
+import static java.sql.Types.DOUBLE;
+import static java.sql.Types.FLOAT;
+import static java.sql.Types.INTEGER;
+import static java.sql.Types.LONGVARBINARY;
+import static java.sql.Types.NUMERIC;
+import static java.sql.Types.CHAR;
+import static java.sql.Types.DATE;
+import static java.sql.Types.LONGNVARCHAR;
+import static java.sql.Types.LONGVARCHAR;
+import static java.sql.Types.NCHAR;
+import static java.sql.Types.NVARCHAR;
+import static java.sql.Types.REAL;
+import static java.sql.Types.ROWID;
+import static java.sql.Types.SMALLINT;
+import static java.sql.Types.TIME;
+import static java.sql.Types.TIMESTAMP;
+import static java.sql.Types.TINYINT;
+import static java.sql.Types.VARBINARY;
+import static java.sql.Types.VARCHAR;
+
+/**
+ * A base class for common code shared by processors that fetch RDBMS data.
+ */
+public abstract class AbstractDatabaseFetchProcessor extends AbstractSessionFactoryProcessor {
+
+ // Relationships
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("Successfully created FlowFile from SQL query result set.")
+ .build();
+
+ protected Set<Relationship> relationships;
+
+ // Properties
+ public static final PropertyDescriptor DBCP_SERVICE = new PropertyDescriptor.Builder()
+ .name("Database Connection Pooling Service")
+ .description("The Controller Service that is used to obtain a connection to the database.")
+ .required(true)
+ .identifiesControllerService(DBCPService.class)
+ .build();
+
+ public static final PropertyDescriptor TABLE_NAME = new PropertyDescriptor.Builder()
+ .name("Table Name")
+ .description("The name of the database table to be queried.")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor COLUMN_NAMES = new PropertyDescriptor.Builder()
+ .name("Columns to Return")
+ .description("A comma-separated list of column names to be used in the query. If your database requires "
+ + "special treatment of the names (quoting, e.g.), each name should include such treatment. If no "
+ + "column names are supplied, all columns in the specified table will be returned.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor MAX_VALUE_COLUMN_NAMES = new PropertyDescriptor.Builder()
+ .name("Maximum-value Columns")
+ .description("A comma-separated list of column names. The processor will keep track of the maximum value "
--- End diff --
This description should include details regarding what the functionality is when no columns are listed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71074429
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java ---
@@ -0,0 +1,361 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.processor.AbstractSessionFactoryProcessor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.sql.Timestamp;
+import java.text.DecimalFormat;
+import java.text.ParseException;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.ServiceLoader;
+import java.util.Set;
+
+import static java.sql.Types.ARRAY;
+import static java.sql.Types.BIGINT;
+import static java.sql.Types.BINARY;
+import static java.sql.Types.BIT;
+import static java.sql.Types.BLOB;
+import static java.sql.Types.BOOLEAN;
+import static java.sql.Types.CLOB;
+import static java.sql.Types.DECIMAL;
+import static java.sql.Types.DOUBLE;
+import static java.sql.Types.FLOAT;
+import static java.sql.Types.INTEGER;
+import static java.sql.Types.LONGVARBINARY;
+import static java.sql.Types.NUMERIC;
+import static java.sql.Types.CHAR;
+import static java.sql.Types.DATE;
+import static java.sql.Types.LONGNVARCHAR;
+import static java.sql.Types.LONGVARCHAR;
+import static java.sql.Types.NCHAR;
+import static java.sql.Types.NVARCHAR;
+import static java.sql.Types.REAL;
+import static java.sql.Types.ROWID;
+import static java.sql.Types.SMALLINT;
+import static java.sql.Types.TIME;
+import static java.sql.Types.TIMESTAMP;
+import static java.sql.Types.TINYINT;
+import static java.sql.Types.VARBINARY;
+import static java.sql.Types.VARCHAR;
+
+/**
+ * A base class for common code shared by processors that fetch RDBMS data.
+ */
+public abstract class AbstractDatabaseFetchProcessor extends AbstractSessionFactoryProcessor {
+
+ // Relationships
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("Successfully created FlowFile from SQL query result set.")
+ .build();
+
+ protected Set<Relationship> relationships;
+
+ // Properties
+ public static final PropertyDescriptor DBCP_SERVICE = new PropertyDescriptor.Builder()
+ .name("Database Connection Pooling Service")
+ .description("The Controller Service that is used to obtain a connection to the database.")
+ .required(true)
+ .identifiesControllerService(DBCPService.class)
+ .build();
+
+ public static final PropertyDescriptor TABLE_NAME = new PropertyDescriptor.Builder()
+ .name("Table Name")
+ .description("The name of the database table to be queried.")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor COLUMN_NAMES = new PropertyDescriptor.Builder()
+ .name("Columns to Return")
+ .description("A comma-separated list of column names to be used in the query. If your database requires "
+ + "special treatment of the names (quoting, e.g.), each name should include such treatment. If no "
+ + "column names are supplied, all columns in the specified table will be returned.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor MAX_VALUE_COLUMN_NAMES = new PropertyDescriptor.Builder()
+ .name("Maximum-value Columns")
+ .description("A comma-separated list of column names. The processor will keep track of the maximum value "
+ + "for each column that has been returned since the processor started running. This can be used to "
+ + "retrieve only those rows that have been added/updated since the last retrieval. Note that some "
+ + "JDBC types such as bit/boolean are not conducive to maintaining maximum value, so columns of these "
+ + "types should not be listed in this property, and will result in error(s) during processing.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor QUERY_TIMEOUT = new PropertyDescriptor.Builder()
+ .name("Max Wait Time")
+ .description("The maximum amount of time allowed for a running SQL select query "
+ + ", zero means there is no limit. Max time less than 1 second will be equal to zero.")
+ .defaultValue("0 seconds")
+ .required(true)
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .build();
+
+ protected List<PropertyDescriptor> propDescriptors;
+
+ public static final PropertyDescriptor DB_TYPE;
+
+ protected final static Map<String, DatabaseAdapter> dbAdapters = new HashMap<>();
+ protected final Map<String, Integer> columnTypeMap = new HashMap<>();
+
+ static {
+ // Load the DatabaseAdapters
+ ServiceLoader<DatabaseAdapter> dbAdapterLoader = ServiceLoader.load(DatabaseAdapter.class);
+ dbAdapterLoader.forEach(it -> dbAdapters.put(it.getName(), it));
+
+ DB_TYPE = new PropertyDescriptor.Builder()
+ .name("db-fetch-db-type")
+ .displayName("Database Type")
+ .description("The type/flavor of database, used for generating database-specific code. In many cases the Generic type "
+ + "should suffice, but some databases (such as Oracle) require custom SQL clauses. ")
+ .allowableValues(dbAdapters.keySet())
+ .defaultValue(dbAdapters.values().stream().findFirst().get().getName())
+ .required(true)
+ .build();
+ }
+
+ public void setup(final ProcessContext context) {
+ // Try to fill the columnTypeMap with the types of the desired max-value columns
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ // Try a query that returns no rows, for the purposes of getting metadata about the columns. It is possible
+ // to use DatabaseMetaData.getColumns(), but not all drivers support this, notably the schema-on-read
+ // approach as in Apache Drill
+ String query = dbAdapter.getSelectStatement(tableName, maxValueColumnNames, "1 = 0", null, null, null);
+ ResultSet resultSet = st.executeQuery(query);
+ ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
+ int numCols = resultSetMetaData.getColumnCount();
+ if (numCols > 0) {
+ columnTypeMap.clear();
+ for (int i = 1; i <= numCols; i++) {
+ String colName = resultSetMetaData.getColumnName(i).toLowerCase();
+ int colType = resultSetMetaData.getColumnType(i);
+ columnTypeMap.put(colName, colType);
+ }
+
+ } else {
+ throw new ProcessException("No columns found in table from those specified: " + maxValueColumnNames);
+ }
+
+ } catch (SQLException e) {
+ throw new ProcessException("Unable to communicate with database in order to determine column types", e);
+ }
+ }
+
+ protected static String getMaxValueFromRow(ResultSet resultSet,
+ int columnIndex,
+ Integer type,
+ String maxValueString,
+ String preProcessStrategy)
--- End diff --
The last parameter shouldn't be "preProcessStrategy" since that property was removed in favor of database type.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073744
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
+
+
+ public static final PropertyDescriptor PARTITION_SIZE = new PropertyDescriptor.Builder()
+ .name("gen-table-fetch-partition-size")
+ .displayName("Partition Size")
+ .description("The number of result rows to be fetched by each generated SQL statement. The total number of rows in "
+ + "the table divided by the partition size gives the number of SQL statements (i.e. FlowFiles) generated. A "
+ + "value of zero indicates that a single FlowFile is to be generated whose SQL statement will fetch all rows "
+ + "in the table.")
+ .defaultValue("10000")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .build();
+
+ public GenerateTableFetch() {
+ final Set<Relationship> r = new HashSet<>();
+ r.add(REL_SUCCESS);
+ relationships = Collections.unmodifiableSet(r);
+
+ final List<PropertyDescriptor> pds = new ArrayList<>();
+ pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
+ pds.add(TABLE_NAME);
+ pds.add(COLUMN_NAMES);
+ pds.add(MAX_VALUE_COLUMN_NAMES);
+ pds.add(QUERY_TIMEOUT);
+ pds.add(PARTITION_SIZE);
+ propDescriptors = Collections.unmodifiableList(pds);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propDescriptors;
+ }
+
+ @OnScheduled
+ public void setup(final ProcessContext context) {
+ super.setup(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
+ ProcessSession session = sessionFactory.createSession();
+ final ComponentLog logger = getLogger();
+
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String columnNames = context.getProperty(COLUMN_NAMES).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final int partitionSize = context.getProperty(PARTITION_SIZE).asInteger();
+
+ final StateManager stateManager = context.getStateManager();
+ final StateMap stateMap;
+
+ try {
+ stateMap = stateManager.getState(Scope.CLUSTER);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve observed maximum values from the State Manager. Will not perform "
+ + "query until this is accomplished.", ioe);
+ context.yield();
+ return;
+ }
+ // Make a mutable copy of the current state property map. This will be updated by the result row callback, and eventually
+ // set as the current state map (after the session has been committed)
+ final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
+
+ // Build a WHERE clause with maximum-value columns (if they exist), and a list of column names that will contain MAX(<column>) aliases. The
+ // executed SQL query will retrieve the count of all records after the filter(s) have been applied, as well as the new maximum values for the
+ // specified columns. This allows the processor to generate the correctly partitioned SQL statements as well as to update the state with the
+ // latest observed maximum values.
+ String whereClause = null;
+ List<String> maxValueColumnNameList = StringUtils.isEmpty(maxValueColumnNames)
+ ? new ArrayList<>(0)
+ : Arrays.asList(maxValueColumnNames.split("\\s*,\\s*"));
+ List<String> maxValueClauses = new ArrayList<>(maxValueColumnNameList.size());
+
+ String columnsClause = null;
+ List<String> maxValueSelectColumns = new ArrayList<>(maxValueColumnNameList.size() + 1);
+ maxValueSelectColumns.add("COUNT(*)");
+
+ // For each maximum-value column, get a WHERE filter and a MAX(column) alias
+ maxValueColumnNameList.forEach(colName -> {
+ maxValueSelectColumns.add("MAX(" + colName + ") " + colName);
+ String maxValue = statePropertyMap.get(colName.toLowerCase());
+ if (!StringUtils.isEmpty(maxValue)) {
+ Integer type = columnTypeMap.get(colName.toLowerCase());
+ if (type == null) {
+ // This shouldn't happen as we are populating columnTypeMap when the processor is scheduled.
+ throw new IllegalArgumentException("No column type found for: " + colName);
+ }
+ // Add a condition for the WHERE clause
+ maxValueClauses.add(colName + " > " + getLiteralByType(type, maxValue, dbAdapter.getName()));
+ }
+ });
+
+ whereClause = StringUtils.join(maxValueClauses, " AND ");
+ columnsClause = StringUtils.join(maxValueSelectColumns, ", ");
+
+ // Build a SELECT query with maximum-value columns (if present)
+ final String selectQuery = dbAdapter.getSelectStatement(tableName, columnsClause, whereClause, null, null, null);
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).asTimePeriod(TimeUnit.SECONDS).intValue();
+ st.setQueryTimeout(queryTimeout); // timeout in seconds
+
+ logger.debug("Executing {}", new Object[]{selectQuery});
+ ResultSet resultSet = st.executeQuery(selectQuery);
+ if (resultSet.next()) {
+ // Total row count is in the first column
+ int rowCount = resultSet.getInt(1);
+
+ // Update the state map with the newly-observed maximum values
+ ResultSetMetaData rsmd = resultSet.getMetaData();
+ for (int i = 2; i <= rsmd.getColumnCount(); i++) {
+ String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ int type = rsmd.getColumnType(i);
+ try {
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ if (newMaxValue != null) {
+ statePropertyMap.put(resultColumnName, newMaxValue);
--- End diff --
This was dumb, this just puts it in the map and line 244 adds it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by JPercivall <gi...@git.apache.org>.
Github user JPercivall commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073287
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
--- End diff --
Does it make sense to allow for multiple onTrigger calls at the same time? Maybe add the "@TriggerSerially" annotation?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #645: NIFI-2157: Add GenerateTableFetch processor
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/645#discussion_r71073742
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -0,0 +1,254 @@
+/*
+ * 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.nifi.processors.standard;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.EventDriven;
+import org.apache.nifi.annotation.behavior.InputRequirement;
+import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
+import org.apache.nifi.annotation.behavior.Stateful;
+import org.apache.nifi.annotation.behavior.WritesAttribute;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.state.Scope;
+import org.apache.nifi.components.state.StateManager;
+import org.apache.nifi.components.state.StateMap;
+import org.apache.nifi.dbcp.DBCPService;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.logging.ComponentLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.text.ParseException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+
+@EventDriven
+@InputRequirement(Requirement.INPUT_FORBIDDEN)
+@Tags({"sql", "select", "jdbc", "query", "database", "fetch", "generate"})
+@SeeAlso(QueryDatabaseTable.class)
+@CapabilityDescription("Generates SQL select queries that fetch \"pages\" of rows from a table. The partition size property, along with the table's row count, "
+ + "determine the size and number of pages and generated FlowFiles. In addition, incremental fetching can be achieved by setting Maximum-Value Columns, "
+ + "which causes the processor to track the columns' maximum values, thus only fetching rows whose columns' values exceed the observed maximums.")
+@Stateful(scopes = Scope.CLUSTER, description = "After performing a query on the specified table, the maximum values for "
+ + "the specified column(s) will be retained for use in future executions of the query. This allows the Processor "
+ + "to fetch only those records that have max values greater than the retained values. This can be used for "
+ + "incremental fetching, fetching of newly added rows, etc. To clear the maximum values, clear the state of the processor "
+ + "per the State Management documentation")
+@WritesAttribute(attribute = "querydbtable.row.count")
+public class GenerateTableFetch extends AbstractDatabaseFetchProcessor {
+
+
+ public static final PropertyDescriptor PARTITION_SIZE = new PropertyDescriptor.Builder()
+ .name("gen-table-fetch-partition-size")
+ .displayName("Partition Size")
+ .description("The number of result rows to be fetched by each generated SQL statement. The total number of rows in "
+ + "the table divided by the partition size gives the number of SQL statements (i.e. FlowFiles) generated. A "
+ + "value of zero indicates that a single FlowFile is to be generated whose SQL statement will fetch all rows "
+ + "in the table.")
+ .defaultValue("10000")
+ .required(true)
+ .expressionLanguageSupported(false)
+ .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .build();
+
+ public GenerateTableFetch() {
+ final Set<Relationship> r = new HashSet<>();
+ r.add(REL_SUCCESS);
+ relationships = Collections.unmodifiableSet(r);
+
+ final List<PropertyDescriptor> pds = new ArrayList<>();
+ pds.add(DBCP_SERVICE);
+ pds.add(DB_TYPE);
+ pds.add(TABLE_NAME);
+ pds.add(COLUMN_NAMES);
+ pds.add(MAX_VALUE_COLUMN_NAMES);
+ pds.add(QUERY_TIMEOUT);
+ pds.add(PARTITION_SIZE);
+ propDescriptors = Collections.unmodifiableList(pds);
+ }
+
+ @Override
+ public Set<Relationship> getRelationships() {
+ return relationships;
+ }
+
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return propDescriptors;
+ }
+
+ @OnScheduled
+ public void setup(final ProcessContext context) {
+ super.setup(context);
+ }
+
+ @Override
+ public void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory) throws ProcessException {
+ ProcessSession session = sessionFactory.createSession();
+ final ComponentLog logger = getLogger();
+
+ final DBCPService dbcpService = context.getProperty(DBCP_SERVICE).asControllerService(DBCPService.class);
+ final DatabaseAdapter dbAdapter = dbAdapters.get(context.getProperty(DB_TYPE).getValue());
+ final String tableName = context.getProperty(TABLE_NAME).getValue();
+ final String columnNames = context.getProperty(COLUMN_NAMES).getValue();
+ final String maxValueColumnNames = context.getProperty(MAX_VALUE_COLUMN_NAMES).getValue();
+ final int partitionSize = context.getProperty(PARTITION_SIZE).asInteger();
+
+ final StateManager stateManager = context.getStateManager();
+ final StateMap stateMap;
+
+ try {
+ stateMap = stateManager.getState(Scope.CLUSTER);
+ } catch (final IOException ioe) {
+ getLogger().error("Failed to retrieve observed maximum values from the State Manager. Will not perform "
+ + "query until this is accomplished.", ioe);
+ context.yield();
+ return;
+ }
+ // Make a mutable copy of the current state property map. This will be updated by the result row callback, and eventually
+ // set as the current state map (after the session has been committed)
+ final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
+
+ // Build a WHERE clause with maximum-value columns (if they exist), and a list of column names that will contain MAX(<column>) aliases. The
+ // executed SQL query will retrieve the count of all records after the filter(s) have been applied, as well as the new maximum values for the
+ // specified columns. This allows the processor to generate the correctly partitioned SQL statements as well as to update the state with the
+ // latest observed maximum values.
+ String whereClause = null;
+ List<String> maxValueColumnNameList = StringUtils.isEmpty(maxValueColumnNames)
+ ? new ArrayList<>(0)
+ : Arrays.asList(maxValueColumnNames.split("\\s*,\\s*"));
+ List<String> maxValueClauses = new ArrayList<>(maxValueColumnNameList.size());
+
+ String columnsClause = null;
+ List<String> maxValueSelectColumns = new ArrayList<>(maxValueColumnNameList.size() + 1);
+ maxValueSelectColumns.add("COUNT(*)");
+
+ // For each maximum-value column, get a WHERE filter and a MAX(column) alias
+ maxValueColumnNameList.forEach(colName -> {
+ maxValueSelectColumns.add("MAX(" + colName + ") " + colName);
+ String maxValue = statePropertyMap.get(colName.toLowerCase());
+ if (!StringUtils.isEmpty(maxValue)) {
+ Integer type = columnTypeMap.get(colName.toLowerCase());
+ if (type == null) {
+ // This shouldn't happen as we are populating columnTypeMap when the processor is scheduled.
+ throw new IllegalArgumentException("No column type found for: " + colName);
+ }
+ // Add a condition for the WHERE clause
+ maxValueClauses.add(colName + " > " + getLiteralByType(type, maxValue, dbAdapter.getName()));
+ }
+ });
+
+ whereClause = StringUtils.join(maxValueClauses, " AND ");
+ columnsClause = StringUtils.join(maxValueSelectColumns, ", ");
+
+ // Build a SELECT query with maximum-value columns (if present)
+ final String selectQuery = dbAdapter.getSelectStatement(tableName, columnsClause, whereClause, null, null, null);
+
+ try (final Connection con = dbcpService.getConnection();
+ final Statement st = con.createStatement()) {
+
+ final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).asTimePeriod(TimeUnit.SECONDS).intValue();
+ st.setQueryTimeout(queryTimeout); // timeout in seconds
+
+ logger.debug("Executing {}", new Object[]{selectQuery});
+ ResultSet resultSet = st.executeQuery(selectQuery);
+ if (resultSet.next()) {
+ // Total row count is in the first column
+ int rowCount = resultSet.getInt(1);
+
+ // Update the state map with the newly-observed maximum values
+ ResultSetMetaData rsmd = resultSet.getMetaData();
+ for (int i = 2; i <= rsmd.getColumnCount(); i++) {
+ String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ int type = rsmd.getColumnType(i);
+ try {
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ if (newMaxValue != null) {
+ statePropertyMap.put(resultColumnName, newMaxValue);
--- End diff --
statePropertyMap is a mutable copy of the original state map, it's not applied to State until line 244 (after commit)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---