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 2017/01/09 18:45:14 UTC
[GitHub] nifi pull request #1407: NIFI-2881: Added EL support to DB Fetch processors,...
GitHub user mattyb149 opened a pull request:
https://github.com/apache/nifi/pull/1407
NIFI-2881: Added EL support to DB Fetch processors, allow incoming flowfiles to GenerateTableFetch
Thank you for submitting a contribution to Apache NiFi.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [x] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [x] Does your PR title start with NIFI-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [x] Has your PR been rebased against the latest commit within the target branch (typically master)?
- [x] Is your initial contribution a single, squashed commit?
### For code changes:
- [x] Have you ensured that the full suite of tests is executed via mvn -Pcontrib-check clean install at the root nifi folder?
- [x] Have you written or updated unit tests to verify your changes?
- [x] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [x] If applicable, have you updated the LICENSE file, including the main LICENSE file under nifi-assembly?
- [x] If applicable, have you updated the NOTICE file, including the main NOTICE file found under nifi-assembly?
- [x] If adding new Properties, have you added .displayName in addition to .name (programmatic access) for each of the new properties?
### For documentation related changes:
- [x] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/mattyb149/nifi NIFI-2881
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi/pull/1407.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 #1407
----
commit a159feb8a1e753fbadab34b893f5f83f69cac730
Author: Matt Burgess <ma...@apache.org>
Date: 2017-01-09T18:37:03Z
NIFI-2881: Added EL support to DB Fetch processors, allow incoming flowfiles to GenerateTableFetch
----
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors, allow ...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on the issue:
https://github.com/apache/nifi/pull/1407
@mattyb149 Thanks for the update. Sorry for the late response, somehow the notification was slipped through my attention.. I'm going to review it and comment soon!
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98898226
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -212,9 +215,21 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
//If an initial max value for column(s) has been specified using properties, and this column is not in the state manager, sync them to the state property map
- for(final Map.Entry<String,String> maxProp : maxValueProperties.entrySet()){
- if (!statePropertyMap.containsKey(maxProp.getKey().toLowerCase())) {
- statePropertyMap.put(maxProp.getKey().toLowerCase(), maxProp.getValue());
+ for (final Map.Entry<String, String> maxProp : maxValueProperties.entrySet()) {
+ String maxPropKey = maxProp.getKey().toLowerCase();
+ String fullyQualifiedMaxPropKey = getStateKey(tableName, maxPropKey);
+ if (!statePropertyMap.containsKey(fullyQualifiedMaxPropKey)) {
+ String newMaxPropValue;
+ // If we can't find the value at the fully-qualified key name, it is possible (under a previous scheme)
+ // the value has been stored under a key that is only the column name. Fall back to check the column name,
+ // but store the new initial max value under the fully-qualified key.
+ if (statePropertyMap.containsKey(maxPropKey)) {
+ newMaxPropValue = statePropertyMap.get(maxPropKey);
+ } else {
+ newMaxPropValue = maxProp.getValue();
--- End diff --
Agreed, I was faced with the choice of 1) an extra query for QueryDatabaseTable, which might happen when scheduled (if using static table names and max-value columns) or once for each table, or 2) not allowing incoming flow files for QDT. I chose option 2 for the reason you point out, the other processors offer that functionality in a more robust manner IMHO. It sounds like we're on the same page, I will add @SeeAlso references to the processors you mention, along with some more text in the CapabilityDescription for QDT to refer to the other processors.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r95799167
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -127,6 +129,7 @@
.defaultValue("0")
.required(true)
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(true)
.build();
public QueryDatabaseTable() {
--- End diff --
I agree that Initial Max Value would be useful for GenerateTableFetch as well. https://issues.apache.org/jira/browse/NIFI-2583 was written and implemented by a community member, his focus was on QueryDatabaseTable at the time. Jira is down ATM but I will write an improvement Jira for GenerateTableFetch and post the link here.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi/pull/1407
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r95627157
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -115,20 +128,36 @@ public GenerateTableFetch() {
@OnScheduled
public void setup(final ProcessContext context) {
+ // The processor is invalid if there is an incoming connection and max-value columns are defined
+ if (context.getProperty(MAX_VALUE_COLUMN_NAMES).isSet() && context.hasIncomingConnection()) {
+ throw new ProcessException("If an incoming connection is supplied, no max-value column names may be specified");
--- End diff --
The original concern is for tables that don't share the same name for max-value column. Now that you mention it, I agree that we could allow for a static string or EL result (maybe without using an incoming flow file?), and just document that all the specified tables must contain the max-value columns. This makes it consistent with Column Names, which currently doesn't have that check but has the same requirement. I will add doc to both and update the logic, thanks!
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r95801717
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -115,20 +128,36 @@ public GenerateTableFetch() {
@OnScheduled
public void setup(final ProcessContext context) {
+ // The processor is invalid if there is an incoming connection and max-value columns are defined
+ if (context.getProperty(MAX_VALUE_COLUMN_NAMES).isSet() && context.hasIncomingConnection()) {
+ throw new ProcessException("If an incoming connection is supplied, no max-value column names may be specified");
--- End diff --
I thought about supporting the older format, but that could lead to problems depending on which table name you pass in. Using your "users" and "purchase_histories" tables above, let's say I was running with the old version and a hard-coded "purchase_histories" table, which stores "last_updated" in the state map. Then with the new version the first table name I pass in via an attribute is "users". I will not find "users.last_updated" so I would check for just "last_updated", whose value is not associated with the users table but rather the purchase_histories table. This is an edge case but I would hate to see the very first run of the processor fail when it used to work.
I do like the example you have of a mapping of tables to max-value columns, I think other products (GoldenGate or Sqoop or something?) allows for this flexibility (you just have to provide your own map). If we end up supporting Max-Value columns with incoming connections then I will make sure this capability is present.
I'm most worried about the arbitrary number of entries in the state map. Once the total size gets above 1MB, I think ZooKeeper starts acting strangely and I certainly wouldn't want this processor to affect the entire NiFi system. This too is an edge case, since I imagine most entries are small (~64 bytes max?) so it would only happen if the number of tables/columns was very large, or for some reason the max-values were large. Historically I've seen limitations placed on other NiFi resources (threads, e.g.) to ensure a discrete maximum to avoid the issues due to arbitrarily large things.
I would very much like to allow for the Max-Value columns, do you have any suggestions on how the processor should behave in the face of an arbitrarily large state map? Perhaps we could set an artificial limit on number of entries (implying a limit on the number of table/columns) and route future flow files (whose table is not yet present in the state map) to a "state map full" relationship or something.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98680344
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -202,40 +246,60 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
ResultSetMetaData rsmd = resultSet.getMetaData();
for (int i = 2; i <= rsmd.getColumnCount(); i++) {
String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ String fullyQualifiedStateKey = getStateKey(tableName, resultColumnName);
+ String resultColumnCurrentMax = statePropertyMap.get(fullyQualifiedStateKey);
+ if (StringUtils.isEmpty(resultColumnCurrentMax) && !isDynamicTableName) {
+ // If we can't find the value at the fully-qualified key name and the table name is static, it is possible (under a previous scheme)
+ // the value has been stored under a key that is only the column name. Fall back to check the column name; either way, when a new
+ // maximum value is observed, it will be stored under the fully-qualified key from then on.
+ resultColumnCurrentMax = statePropertyMap.get(resultColumnName);
+ }
+
int type = rsmd.getColumnType(i);
+ if (isDynamicTableName) {
+ // We haven't pre-populated the column type map if the table name is dynamic, so do it here
+ columnTypeMap.put(fullyQualifiedStateKey, type);
+ }
try {
- String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, resultColumnCurrentMax, dbAdapter.getName());
if (newMaxValue != null) {
- statePropertyMap.put(resultColumnName, newMaxValue);
+ statePropertyMap.put(fullyQualifiedStateKey, 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);
+ 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;
+ // Generate SQL statements to read "pages" of data
+ for (int i = 0; i < numberOfFetches; i++) {
+ Integer limit = partitionSize == 0 ? null : partitionSize;
+ Integer offset = partitionSize == 0 ? null : i * partitionSize;
+ final String query = dbAdapter.getSelectStatement(tableName, columnNames, whereClause, StringUtils.join(maxValueColumnNameList, ", "), limit, offset);
+ FlowFile sqlFlowFile = (fileToProcess == null) ? session.create() : session.create(fileToProcess);
+ sqlFlowFile = session.write(sqlFlowFile, out -> out.write(query.getBytes()));
+ session.transfer(sqlFlowFile, REL_SUCCESS);
+ }
+
+ if (fileToProcess != null) {
+ session.remove(fileToProcess);
+ }
+ } catch (SQLException e) {
+ if (fileToProcess != null) {
+ logger.error("Unable to execute SQL select query {} due to {}, routing {} to failure", new Object[]{selectQuery, e, fileToProcess});
+ session.transfer(fileToProcess, REL_FAILURE);
--- End diff --
agreed, will add the message (not the stack trace) as an attribute.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98604895
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -202,40 +246,60 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
ResultSetMetaData rsmd = resultSet.getMetaData();
for (int i = 2; i <= rsmd.getColumnCount(); i++) {
String resultColumnName = rsmd.getColumnName(i).toLowerCase();
+ String fullyQualifiedStateKey = getStateKey(tableName, resultColumnName);
+ String resultColumnCurrentMax = statePropertyMap.get(fullyQualifiedStateKey);
+ if (StringUtils.isEmpty(resultColumnCurrentMax) && !isDynamicTableName) {
+ // If we can't find the value at the fully-qualified key name and the table name is static, it is possible (under a previous scheme)
+ // the value has been stored under a key that is only the column name. Fall back to check the column name; either way, when a new
+ // maximum value is observed, it will be stored under the fully-qualified key from then on.
+ resultColumnCurrentMax = statePropertyMap.get(resultColumnName);
+ }
+
int type = rsmd.getColumnType(i);
+ if (isDynamicTableName) {
+ // We haven't pre-populated the column type map if the table name is dynamic, so do it here
+ columnTypeMap.put(fullyQualifiedStateKey, type);
+ }
try {
- String newMaxValue = getMaxValueFromRow(resultSet, i, type, statePropertyMap.get(resultColumnName.toLowerCase()), dbAdapter.getName());
+ String newMaxValue = getMaxValueFromRow(resultSet, i, type, resultColumnCurrentMax, dbAdapter.getName());
if (newMaxValue != null) {
- statePropertyMap.put(resultColumnName, newMaxValue);
+ statePropertyMap.put(fullyQualifiedStateKey, 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);
+ 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;
+ // Generate SQL statements to read "pages" of data
+ for (int i = 0; i < numberOfFetches; i++) {
+ Integer limit = partitionSize == 0 ? null : partitionSize;
+ Integer offset = partitionSize == 0 ? null : i * partitionSize;
+ final String query = dbAdapter.getSelectStatement(tableName, columnNames, whereClause, StringUtils.join(maxValueColumnNameList, ", "), limit, offset);
+ FlowFile sqlFlowFile = (fileToProcess == null) ? session.create() : session.create(fileToProcess);
+ sqlFlowFile = session.write(sqlFlowFile, out -> out.write(query.getBytes()));
+ session.transfer(sqlFlowFile, REL_SUCCESS);
+ }
+
+ if (fileToProcess != null) {
+ session.remove(fileToProcess);
+ }
+ } catch (SQLException e) {
+ if (fileToProcess != null) {
+ logger.error("Unable to execute SQL select query {} due to {}, routing {} to failure", new Object[]{selectQuery, e, fileToProcess});
+ session.transfer(fileToProcess, REL_FAILURE);
--- End diff --
Embedding Exception detail to an attribute might be helpful.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98810570
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -212,9 +215,21 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
//If an initial max value for column(s) has been specified using properties, and this column is not in the state manager, sync them to the state property map
- for(final Map.Entry<String,String> maxProp : maxValueProperties.entrySet()){
- if (!statePropertyMap.containsKey(maxProp.getKey().toLowerCase())) {
- statePropertyMap.put(maxProp.getKey().toLowerCase(), maxProp.getValue());
+ for (final Map.Entry<String, String> maxProp : maxValueProperties.entrySet()) {
+ String maxPropKey = maxProp.getKey().toLowerCase();
+ String fullyQualifiedMaxPropKey = getStateKey(tableName, maxPropKey);
+ if (!statePropertyMap.containsKey(fullyQualifiedMaxPropKey)) {
+ String newMaxPropValue;
+ // If we can't find the value at the fully-qualified key name, it is possible (under a previous scheme)
+ // the value has been stored under a key that is only the column name. Fall back to check the column name,
+ // but store the new initial max value under the fully-qualified key.
+ if (statePropertyMap.containsKey(maxPropKey)) {
+ newMaxPropValue = statePropertyMap.get(maxPropKey);
+ } else {
+ newMaxPropValue = maxProp.getValue();
--- End diff --
I was suggesting that for QueryDatabaseTable and Initial max value, instead of GenerateTableFetch. Because not being able to know the column type seemed an obstacle to support incoming Flow Files in QueryDatabaseTable ([PR comment](https://github.com/apache/nifi/pull/1407#issuecomment-274592165) and [JIRA comment](https://issues.apache.org/jira/browse/NIFI-2881?focusedCommentId=15811850&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-15811850)).
However, I agree with you that, as you've written, since GenerateTableFetch and ExecuteSQL can be used if user need to pass arguments from incoming flow files, it's not necessary to support incoming flow files in QueryDatabaseTable.
If we are not going to add incoming flow file support for QueryDatabaseTable in this PR, maybe we should add some docs in QueryDatabaseTable capability description to guide users to refer GenerateTableFetch and ExecuteSQL, 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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r95719637
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -127,6 +129,7 @@
.defaultValue("0")
.required(true)
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(true)
.build();
public QueryDatabaseTable() {
--- End diff --
This is just a question out of curiosity, QueryDatabaseTable supports dynamic properties `Initial Max Value`, but GenerateTableFetch doesn't. As my understanding, QueryDatabaseTable and GenerateTableFetch are identical except the former execute SQL query while the latter generate flow files containing SQL. Initial Max Value dynamic property would be useful for GenerateTableFetch, too.
Also wondering how it will work if table name is specified by EL.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r99103861
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -212,9 +215,21 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
//If an initial max value for column(s) has been specified using properties, and this column is not in the state manager, sync them to the state property map
- for(final Map.Entry<String,String> maxProp : maxValueProperties.entrySet()){
- if (!statePropertyMap.containsKey(maxProp.getKey().toLowerCase())) {
- statePropertyMap.put(maxProp.getKey().toLowerCase(), maxProp.getValue());
+ for (final Map.Entry<String, String> maxProp : maxValueProperties.entrySet()) {
+ String maxPropKey = maxProp.getKey().toLowerCase();
+ String fullyQualifiedMaxPropKey = getStateKey(tableName, maxPropKey);
+ if (!statePropertyMap.containsKey(fullyQualifiedMaxPropKey)) {
+ String newMaxPropValue;
+ // If we can't find the value at the fully-qualified key name, it is possible (under a previous scheme)
+ // the value has been stored under a key that is only the column name. Fall back to check the column name,
+ // but store the new initial max value under the fully-qualified key.
+ if (statePropertyMap.containsKey(maxPropKey)) {
+ newMaxPropValue = statePropertyMap.get(maxPropKey);
+ } else {
+ newMaxPropValue = maxProp.getValue();
--- End diff --
Thanks for updating the docs, it became clearer :)
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors, allow ...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on the issue:
https://github.com/apache/nifi/pull/1407
Thanks @mattyb149 , LGTM! I've written up a Gist [NiFi Example: Change detection from Multiple Tables](https://gist.github.com/ijokarumawak/ed9085024eeeefbca19cfb2f20d23ed4) with sample template and table DDL that I used for testing this.
I'm going to squash commits and merge it to master.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98684537
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -212,9 +215,21 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
//If an initial max value for column(s) has been specified using properties, and this column is not in the state manager, sync them to the state property map
- for(final Map.Entry<String,String> maxProp : maxValueProperties.entrySet()){
- if (!statePropertyMap.containsKey(maxProp.getKey().toLowerCase())) {
- statePropertyMap.put(maxProp.getKey().toLowerCase(), maxProp.getValue());
+ for (final Map.Entry<String, String> maxProp : maxValueProperties.entrySet()) {
+ String maxPropKey = maxProp.getKey().toLowerCase();
+ String fullyQualifiedMaxPropKey = getStateKey(tableName, maxPropKey);
+ if (!statePropertyMap.containsKey(fullyQualifiedMaxPropKey)) {
+ String newMaxPropValue;
+ // If we can't find the value at the fully-qualified key name, it is possible (under a previous scheme)
+ // the value has been stored under a key that is only the column name. Fall back to check the column name,
+ // but store the new initial max value under the fully-qualified key.
+ if (statePropertyMap.containsKey(maxPropKey)) {
+ newMaxPropValue = statePropertyMap.get(maxPropKey);
+ } else {
+ newMaxPropValue = maxProp.getValue();
--- End diff --
I was trying to limit the number of queries GenerateTableFetch has to make, since the real work will be done downstream. In this case, an invariant is that a columnTypeMap entry exists iff a max-value entry exists. Otherwise the first time I'll be doing a SELECT MAX(max-value-column) without a WHERE clause (since there is no max-value entry) so I don't have to worry about type conversions for literals. The type of the value(s) is the same as the type of the MAX(values), so the first time I can fill both the columnType map and the max-value map.
Is there a path where the current approach does not suffice? If so then I can certainly add an extra query.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98604070
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java ---
@@ -117,9 +124,11 @@
+ "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. If no columns "
- + "are provided, all rows from the table will be considered, which could have a performance impact.")
+ + "are provided, all rows from the table will be considered, which could have a performance impact.\nNOTE: If Expression Language is "
+ + "present for this property and it refers to flow file attribute(s), then the Table Name property must also contain Expression Language.")
--- End diff --
What does the 'NOTE' imply? There's no validation code for this condition, so I understand it as a recommendation. Is it correct?
I think it's possible that an user intentionally use static Table name and dynamic Max value columns together, if they would like to fetch `brand new` items by `created_timestamp`, and also `updated` items by `updated_timestamp` from the same table. This works fine.
What doesn't work (or dangerous) is probably a situation that different Max value columns are passed with the same table name and those columns overlap each other, as such usage will corrupt state management.
- E.g.1 : FF1(table=T1, maxColumn=b, a) but FF2(table=T1, maxColumn=a only). This can be problematic.
- E.g.2 : FF1(table=T1, maxColumn=b, a), FF2(table=T1, maxColumn=c). This actually work fine.
If the note was added to explain this risk, then the 'NOTE' can be written like: "NOTE: It's important to use consistent Max value columns for a given table for increment fetch to work properly"?
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r99111299
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java ---
@@ -117,9 +124,11 @@
+ "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. If no columns "
- + "are provided, all rows from the table will be considered, which could have a performance impact.")
+ + "are provided, all rows from the table will be considered, which could have a performance impact.\nNOTE: If Expression Language is "
+ + "present for this property and it refers to flow file attribute(s), then the Table Name property must also contain Expression Language.")
--- End diff --
Shoot, I forgot this one :( will update this time
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98615920
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -212,9 +215,21 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
//If an initial max value for column(s) has been specified using properties, and this column is not in the state manager, sync them to the state property map
- for(final Map.Entry<String,String> maxProp : maxValueProperties.entrySet()){
- if (!statePropertyMap.containsKey(maxProp.getKey().toLowerCase())) {
- statePropertyMap.put(maxProp.getKey().toLowerCase(), maxProp.getValue());
+ for (final Map.Entry<String, String> maxProp : maxValueProperties.entrySet()) {
--- End diff --
When we add incoming flow file support, it maybe useful if Initial max values can be specified via flow file attributes. Because if table name is dynamic, user won't be able to configure dynamic processor attribute in advance for each possible table.
Example Flow File (table.name='table-1', maxvalue.columns='update_date', initial.maxvalue.update_date='20170101')
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98680239
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java ---
@@ -117,9 +124,11 @@
+ "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. If no columns "
- + "are provided, all rows from the table will be considered, which could have a performance impact.")
+ + "are provided, all rows from the table will be considered, which could have a performance impact.\nNOTE: If Expression Language is "
+ + "present for this property and it refers to flow file attribute(s), then the Table Name property must also contain Expression Language.")
--- End diff --
Good point, I will update the note.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r95716780
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -115,20 +128,36 @@ public GenerateTableFetch() {
@OnScheduled
public void setup(final ProcessContext context) {
+ // The processor is invalid if there is an incoming connection and max-value columns are defined
+ if (context.getProperty(MAX_VALUE_COLUMN_NAMES).isSet() && context.hasIncomingConnection()) {
+ throw new ProcessException("If an incoming connection is supplied, no max-value column names may be specified");
--- End diff --
I understand the concerns.
For backward compatibility, I think we should provide that so that existing flow can keep fetching rows based on the stored state even after upgrade. I've done the similar thing before with [TailFile processor](https://github.com/apache/nifi/blob/master/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/TailFile.java#L348). Checking the state key name format to determine if it's current or older format, then migrate the values.
I concern this statement in your previous comment, If we support Max-value column with incoming files:
> and just document that all the specified tables must contain the max-value columns.
Max-value column hasn't been required, I guess that is for users who want to fetch all rows periodically and don't have to track the max value. Maybe for things like master configuration tables. Then I think we need to keep supporting empty max-value column.
An example flow I thought that might be useful is, using GenerateFlowFile or FetchFile to pass a configuration text such as:
```
# Table name : MAX value column(s)
USERS:LAST_UPDATED
ITEMS
PURCHASE_HISTORIES:LAST_UPDATED
```
Then pass it to SplitText and ExtractText to generate flow files with attributes `tableName` and `maxColumns`. Then pass it to GenerateTableFetch processor to generate fetch SQL dynamically. This way, user can easily modify which table to fetch.
Maybe after processing these incoming flow files, GenerateTableFetch would have state like this (Table `ITEMS` doesn't have max value column):
|KEY|VALUE|
|----|-------|
|USERS.LAST_UPDATED|2017.01.12 11:42:00|
|PURCHASE_HISTORIES.LAST_UPDATED|2017.01.12 11:59:32|
How do you think? Thanks!
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r95557680
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -115,20 +128,36 @@ public GenerateTableFetch() {
@OnScheduled
public void setup(final ProcessContext context) {
+ // The processor is invalid if there is an incoming connection and max-value columns are defined
+ if (context.getProperty(MAX_VALUE_COLUMN_NAMES).isSet() && context.hasIncomingConnection()) {
+ throw new ProcessException("If an incoming connection is supplied, no max-value column names may be specified");
--- End diff --
I got this error while passing incoming flow files to dynamically specify target table name to fetch. But since those target tables has the same column structure, I specified a column name `ID` as `Maximum-value Columns`. What concern is there behind this check condition? I think it'd be useful to specify MAX_VALUE_COLUMN_NAMES by either static string or EL result.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r95629482
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -115,20 +128,36 @@ public GenerateTableFetch() {
@OnScheduled
public void setup(final ProcessContext context) {
+ // The processor is invalid if there is an incoming connection and max-value columns are defined
+ if (context.getProperty(MAX_VALUE_COLUMN_NAMES).isSet() && context.hasIncomingConnection()) {
+ throw new ProcessException("If an incoming connection is supplied, no max-value column names may be specified");
--- End diff --
The other concern was if we support Max Value columns using attributes, we'd have to keep state for those columns for each table, which could get unwieldy for state management depending on the number of columns and tables. Also if I add this logic, I'll need to change how state is stored (to change the key to a full-qualified column name), which could impact backwards compatibility (if state is maintained across an upgrade).
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r95919431
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -115,20 +128,36 @@ public GenerateTableFetch() {
@OnScheduled
public void setup(final ProcessContext context) {
+ // The processor is invalid if there is an incoming connection and max-value columns are defined
+ if (context.getProperty(MAX_VALUE_COLUMN_NAMES).isSet() && context.hasIncomingConnection()) {
+ throw new ProcessException("If an incoming connection is supplied, no max-value column names may be specified");
--- End diff --
That is a good point, for supporting older format and migration. However, the same problem exists even for now. If the processor was configured to fetch from `users` using `last_updated` as max column and ran. The processor has `last_updated` state. Then the user may change table name to `purchase_histories`. Since the processor doesn't implement `onPropertyModified` method to handle these change, I guess the processor will use the state that was actually for different table.
Maybe we can implement something intelligent by capturing the old configuration at onPropertyModified. It maybe a bit difficult though, since we can't access state manager at onPropertyModified method.
For the size of state map, there is a check at [ZooKeeperStateProvider](https://github.com/apache/nifi/blob/master/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/state/providers/zookeeper/ZooKeeperStateProvider.java#L310), so once a state map becomes grater than 1MB in serialized size, the processor would get a specific StateTooLargeException. Although it's too late to rollback the process session because it has already been committed (I think this ordering is correct as it is now, prefer duplicate over loss), we can throw StateTooLargeException so that NiFi framework can yield the processor. Then user can see what went wrong by looking at the bulletin or error log message. Those indicator will keep telling the user until they fix it by for example, split tables to fetch into smaller groups and distribute it to multiple GenerateTableFetch processors to reduce state size.
Since this is an edge case and won't affect other part of flow, and hard to predict the optimal maximum entries for the state, I think throwing StateTooLargeException to framework and yield the processor would be a sufficient handling.
Having `state map full` relationship would be overkill for most case it's unnecessary but it forces user to auto terminate or route to somewhere.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors, allow ...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on the issue:
https://github.com/apache/nifi/pull/1407
If we going to add the ability to keep track of dynamic max-value properties in GenerateTableFetch, then there's really no reason why we can't do the same for QueryDatabaseTable. I left it out of the original Jira based on the premise that we wouldn't support max-value columns with incoming connections, and at that point QueryDatabaseTable would work basically like ExecuteSQL.
The major difference in this case will be that if there is an incoming connection, then the fetch processors are purely event-driven, meaning they will need to be hydrated from upstream to get new results. If there is no incoming connection, then they will maintain their current behavior of running on their current schedule to check for new max values. Does this sound ok?
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98614712
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -212,9 +215,21 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
//If an initial max value for column(s) has been specified using properties, and this column is not in the state manager, sync them to the state property map
- for(final Map.Entry<String,String> maxProp : maxValueProperties.entrySet()){
- if (!statePropertyMap.containsKey(maxProp.getKey().toLowerCase())) {
- statePropertyMap.put(maxProp.getKey().toLowerCase(), maxProp.getValue());
+ for (final Map.Entry<String, String> maxProp : maxValueProperties.entrySet()) {
+ String maxPropKey = maxProp.getKey().toLowerCase();
+ String fullyQualifiedMaxPropKey = getStateKey(tableName, maxPropKey);
+ if (!statePropertyMap.containsKey(fullyQualifiedMaxPropKey)) {
+ String newMaxPropValue;
+ // If we can't find the value at the fully-qualified key name, it is possible (under a previous scheme)
+ // the value has been stored under a key that is only the column name. Fall back to check the column name,
+ // but store the new initial max value under the fully-qualified key.
+ if (statePropertyMap.containsKey(maxPropKey)) {
+ newMaxPropValue = statePropertyMap.get(maxPropKey);
+ } else {
+ newMaxPropValue = maxProp.getValue();
--- End diff --
For the unknown column type issue when initial max value and dynamic max value columns as described in [this comment](#issuecomment-274592165), how about adding a query using "1 = 0" WHERE clause here to populate columnTypeMap if a type for the column doesn't exist in the map?
Since this line can only be achieved at the initial execution for a given table, it should be safe to do so like AbstractDatabaseFetchProcessor.setup() does.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r99103639
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java ---
@@ -117,9 +124,11 @@
+ "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. If no columns "
- + "are provided, all rows from the table will be considered, which could have a performance impact.")
+ + "are provided, all rows from the table will be considered, which could have a performance impact.\nNOTE: If Expression Language is "
+ + "present for this property and it refers to flow file attribute(s), then the Table Name property must also contain Expression Language.")
--- End diff --
Is this going to be updated?
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98681382
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -212,9 +215,21 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
//If an initial max value for column(s) has been specified using properties, and this column is not in the state manager, sync them to the state property map
- for(final Map.Entry<String,String> maxProp : maxValueProperties.entrySet()){
- if (!statePropertyMap.containsKey(maxProp.getKey().toLowerCase())) {
- statePropertyMap.put(maxProp.getKey().toLowerCase(), maxProp.getValue());
+ for (final Map.Entry<String, String> maxProp : maxValueProperties.entrySet()) {
--- End diff --
Agreed. For GenerateTableFetch, initial max values are more useful as flow file attributes, although for static tables / max-value columns (which don't require an incoming connection), it would still be nice to be able to add them as dynamic properties. Do you mind adding your observation(s) to [NIFI-3335](https://issues.apache.org/jira/browse/NIFI-3335)?
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r96434505
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/GenerateTableFetch.java ---
@@ -115,20 +128,36 @@ public GenerateTableFetch() {
@OnScheduled
public void setup(final ProcessContext context) {
+ // The processor is invalid if there is an incoming connection and max-value columns are defined
+ if (context.getProperty(MAX_VALUE_COLUMN_NAMES).isSet() && context.hasIncomingConnection()) {
+ throw new ProcessException("If an incoming connection is supplied, no max-value column names may be specified");
--- End diff --
For backwards-compatibility, what do you think about only falling back to the column name as the key if the table name property does not contain Expression Language? I'm thinking the only use case we should have to support from before is if the table name is static (as it was required to be before now). This would not allow for incoming flow files to specify via attribute the same table name as before, but I don't see that being a high-priority use case necessarily. With this approach, currently configured processors would continue to work as-is, but once you change to incoming flow files (or use EL), the new scheme would apply. Even currently configured processors would automatically switch to the new scheme as soon as a new maximum value is observed.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r95919672
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -127,6 +129,7 @@
.defaultValue("0")
.required(true)
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
+ .expressionLanguageSupported(true)
.build();
public QueryDatabaseTable() {
--- End diff --
Thanks! JIRA is still under maintenance process..
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors,...
Posted by ijokarumawak <gi...@git.apache.org>.
Github user ijokarumawak commented on a diff in the pull request:
https://github.com/apache/nifi/pull/1407#discussion_r98807002
--- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java ---
@@ -212,9 +215,21 @@ public void onTrigger(final ProcessContext context, final ProcessSessionFactory
final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
//If an initial max value for column(s) has been specified using properties, and this column is not in the state manager, sync them to the state property map
- for(final Map.Entry<String,String> maxProp : maxValueProperties.entrySet()){
- if (!statePropertyMap.containsKey(maxProp.getKey().toLowerCase())) {
- statePropertyMap.put(maxProp.getKey().toLowerCase(), maxProp.getValue());
+ for (final Map.Entry<String, String> maxProp : maxValueProperties.entrySet()) {
--- End diff --
Sure, I've updated NIFI-3335 description.
---
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 #1407: NIFI-2881: Added EL support to DB Fetch processors, allow ...
Posted by mattyb149 <gi...@git.apache.org>.
Github user mattyb149 commented on the issue:
https://github.com/apache/nifi/pull/1407
I ran into a snag when trying to add incoming flow file support to QueryDatabaseTable. For GenerateTableFetch, a SQL query is made to the database to get the max values (and column types) for the table. If the table names and/or max-value columns are static (like before), the column types can be fetched when the processor is scheduled. If not, they can be determined from the SQL query results the first time it is issued (no max-values yet, e.g.). However QueryDatabaseTable supports "Initial Max Value" dynamic properties, and does not perform a query ahead of time to get the max values. Instead it performs the actual fetch query, and determines the max value as it iterates over the result set. With an initial max value, the initial query needs to have a WHERE clause, but we don't know the type of the column(s) to determine how to represent the initial max value as a literal.
So as to avoid this issue, I enabled Expression Language support for the QueryDatabaseTable properties but kept the annotation as INPUT_FORBIDDEN. We can revisit this issue as a later improvement to QDT and/or when Initial Max Value is implemented for GenerateTableFetch (as it may have this same limitation at that point).
How does that sound? The updated PR has the aforementioned behavior, both processors support EL (variable-based table names and max-value columns, e.g.) and GenerateTableFetch supports incoming flow files (so attributes can be used in the EL expressions).
---
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.
---