You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mt...@apache.org on 2018/05/11 13:52:08 UTC
nifi git commit: NIFI-5049 Fix handling of Phonenix datetime columns
Repository: nifi
Updated Branches:
refs/heads/master 099bfcdf3 -> 64356e001
NIFI-5049 Fix handling of Phonenix datetime columns
This closes #2625
Signed-off-by: Mike Thomsen <mi...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/64356e00
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/64356e00
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/64356e00
Branch: refs/heads/master
Commit: 64356e001432b78d919604d13787ea7b40e80e8e
Parents: 099bfcd
Author: Gardella Juan Pablo <ga...@gmail.com>
Authored: Wed Apr 11 01:23:00 2018 -0300
Committer: Mike Thomsen <mi...@gmail.com>
Committed: Fri May 11 09:51:24 2018 -0400
----------------------------------------------------------------------
.../AbstractDatabaseFetchProcessor.java | 24 +++++--
.../processors/standard/QueryDatabaseTable.java | 7 +-
.../db/impl/PhoenixDatabaseAdapter.java | 72 ++++++++++++++++++++
.../standard/QueryDatabaseTableTest.java | 38 +++++++++++
4 files changed, 133 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/64356e00/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java
index 0210739..924c7da 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractDatabaseFetchProcessor.java
@@ -30,6 +30,7 @@ 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 org.apache.nifi.processors.standard.db.impl.PhoenixDatabaseAdapter;
import org.apache.nifi.util.StringUtils;
import java.io.IOException;
@@ -485,13 +486,26 @@ public abstract class AbstractDatabaseFetchProcessor extends AbstractSessionFact
case NVARCHAR:
case VARCHAR:
case ROWID:
- case DATE:
- case TIME:
return "'" + value + "'";
+ case TIME:
+ if (PhoenixDatabaseAdapter.NAME.equals(databaseType)) {
+ return "time '" + value + "'";
+ }
+ case DATE:
case TIMESTAMP:
- if (!StringUtils.isEmpty(databaseType) && databaseType.contains("Oracle")) {
- // For backwards compatibility, the type might be TIMESTAMP but the state value is in DATE format. This should be a one-time occurrence as the next maximum value
- // should be stored as a full timestamp. Even so, check to see if the value is missing time-of-day information, and use the "date" coercion rather than the
+ // TODO delegate to database adapter the conversion instead of using if in this
+ // class.
+ // TODO (cont) if a new else is added, please refactor the code.
+ // Ideally we should probably have a method on the adapter to get a clause that
+ // coerces a
+ // column to a Timestamp if need be (the generic one can be a no-op)
+ if (!StringUtils.isEmpty(databaseType)
+ && (databaseType.contains("Oracle") || PhoenixDatabaseAdapter.NAME.equals(databaseType))) {
+ // For backwards compatibility, the type might be TIMESTAMP but the state value
+ // is in DATE format. This should be a one-time occurrence as the next maximum
+ // value
+ // should be stored as a full timestamp. Even so, check to see if the value is
+ // missing time-of-day information, and use the "date" coercion rather than the
// "timestamp" coercion in that case
if (value.matches("\\d{4}-\\d{2}-\\d{2}")) {
return "date '" + value + "'";
http://git-wip-us.apache.org/repos/asf/nifi/blob/64356e00/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java
index 1e8750f..c12b3b9 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryDatabaseTable.java
@@ -310,9 +310,10 @@ public class QueryDatabaseTable extends AbstractDatabaseFetchProcessor {
final Integer queryTimeout = context.getProperty(QUERY_TIMEOUT).evaluateAttributeExpressions().asTimePeriod(TimeUnit.SECONDS).intValue();
st.setQueryTimeout(queryTimeout); // timeout in seconds
- try {
- logger.debug("Executing query {}", new Object[]{selectQuery});
- final ResultSet resultSet = st.executeQuery(selectQuery);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Executing query {}", new Object[] { selectQuery });
+ }
+ try (final ResultSet resultSet = st.executeQuery(selectQuery)) {
int fragmentIndex=0;
while(true) {
final AtomicLong nrOfRows = new AtomicLong(0L);
http://git-wip-us.apache.org/repos/asf/nifi/blob/64356e00/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/PhoenixDatabaseAdapter.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/PhoenixDatabaseAdapter.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/PhoenixDatabaseAdapter.java
new file mode 100644
index 0000000..92522db
--- /dev/null
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/db/impl/PhoenixDatabaseAdapter.java
@@ -0,0 +1,72 @@
+/*
+ * 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.db.impl;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.processors.standard.db.DatabaseAdapter;
+
+/**
+ * A Apache Phoenix database adapter that generates ANSI SQL.
+ */
+public final class PhoenixDatabaseAdapter implements DatabaseAdapter {
+ public static final String NAME = "Phoenix";
+
+ @Override
+ public String getName() {
+ return NAME;
+ }
+
+ @Override
+ public String getDescription() {
+ return "Generates Phoenix compliant SQL";
+ }
+
+ @Override
+ public String getSelectStatement(String tableName, String columnNames, String whereClause, String orderByClause,
+ Long limit, Long offset) {
+ if (StringUtils.isEmpty(tableName)) {
+ throw new IllegalArgumentException("Table name cannot be null or empty");
+ }
+ final StringBuilder query = new StringBuilder("SELECT ");
+ if (StringUtils.isEmpty(columnNames) || columnNames.trim().equals("*")) {
+ query.append("*");
+ } else {
+ query.append(columnNames);
+ }
+ query.append(" FROM ");
+ query.append(tableName);
+
+ if (!StringUtils.isEmpty(whereClause)) {
+ query.append(" WHERE ");
+ query.append(whereClause);
+ }
+ if (!StringUtils.isEmpty(orderByClause)) {
+ query.append(" ORDER BY ");
+ query.append(orderByClause);
+ }
+ if (limit != null) {
+ query.append(" LIMIT ");
+ query.append(limit);
+ }
+ if (offset != null && offset > 0) {
+ query.append(" OFFSET ");
+ query.append(offset);
+ }
+
+ return query.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/64356e00/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/QueryDatabaseTableTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/QueryDatabaseTableTest.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/QueryDatabaseTableTest.java
index ff6a7f0..87bf2a2 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/QueryDatabaseTableTest.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/QueryDatabaseTableTest.java
@@ -32,6 +32,7 @@ import org.apache.nifi.processors.standard.db.impl.GenericDatabaseAdapter;
import org.apache.nifi.processors.standard.db.impl.MSSQLDatabaseAdapter;
import org.apache.nifi.processors.standard.db.impl.MySQLDatabaseAdapter;
import org.apache.nifi.processors.standard.db.impl.OracleDatabaseAdapter;
+import org.apache.nifi.processors.standard.db.impl.PhoenixDatabaseAdapter;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
@@ -180,6 +181,43 @@ public class QueryDatabaseTableTest {
dbAdapter = new OracleDatabaseAdapter();
query = processor.getQuery(dbAdapter, "myTable", null, Arrays.asList("id", "DATE_CREATED"), "type = \"CUSTOMER\"", stateManager.getState(Scope.CLUSTER).toMap());
assertEquals("SELECT * FROM myTable WHERE id > 509 AND DATE_CREATED >= timestamp '2016-03-07 12:34:56' AND (type = \"CUSTOMER\")", query);
+
+ // Test time.
+ processor.putColumnType("mytable" + AbstractDatabaseFetchProcessor.NAMESPACE_DELIMITER + "time_created", Types.TIME);
+ maxValues.clear();
+ maxValues.put("id", "509");
+ maxValues.put("time_created", "12:34:57");
+ maxValues.put("date_created", "2016-03-07 12:34:56");
+ stateManager = runner.getStateManager();
+ stateManager.clear(Scope.CLUSTER);
+ stateManager.setState(maxValues, Scope.CLUSTER);
+ query = processor.getQuery(dbAdapter, "myTable", null, Arrays.asList("id", "DATE_CREATED", "TIME_CREATED"), "type = \"CUSTOMER\"", stateManager.getState(Scope.CLUSTER).toMap());
+ assertEquals("SELECT * FROM myTable WHERE id > 509 AND DATE_CREATED >= timestamp '2016-03-07 12:34:56' AND TIME_CREATED >= timestamp '12:34:57' AND (type = \"CUSTOMER\")", query);
+ dbAdapter = new GenericDatabaseAdapter();
+ query = processor.getQuery(dbAdapter, "myTable", null, Arrays.asList("id", "DATE_CREATED", "TIME_CREATED"), "type = \"CUSTOMER\"", stateManager.getState(Scope.CLUSTER).toMap());
+ assertEquals("SELECT * FROM myTable WHERE id > 509 AND DATE_CREATED >= '2016-03-07 12:34:56' AND TIME_CREATED >= '12:34:57' AND (type = \"CUSTOMER\")", query);
+ }
+
+ @Test
+ public void testGetQueryUsingPhoenixAdapter() throws Exception {
+ Map<String, String> maxValues = new HashMap<>();
+ StateManager stateManager = runner.getStateManager();
+ processor.putColumnType("mytable" + AbstractDatabaseFetchProcessor.NAMESPACE_DELIMITER + "id", Types.INTEGER);
+ processor.putColumnType("mytable" + AbstractDatabaseFetchProcessor.NAMESPACE_DELIMITER + "time_created", Types.TIME);
+ processor.putColumnType("mytable" + AbstractDatabaseFetchProcessor.NAMESPACE_DELIMITER + "date_created", Types.TIMESTAMP);
+
+ maxValues.put("id", "509");
+ maxValues.put("time_created", "12:34:57");
+ maxValues.put("date_created", "2016-03-07 12:34:56");
+ stateManager.setState(maxValues, Scope.CLUSTER);
+
+ dbAdapter = new PhoenixDatabaseAdapter();
+ String query = processor.getQuery(dbAdapter, "myTable", null, Arrays.asList("id", "DATE_CREATED", "TIME_CREATED"), "type = \"CUSTOMER\"", stateManager.getState(Scope.CLUSTER).toMap());
+ assertEquals("SELECT * FROM myTable WHERE id > 509 AND DATE_CREATED >= timestamp '2016-03-07 12:34:56' AND TIME_CREATED >= time '12:34:57' AND (type = \"CUSTOMER\")", query);
+ // Cover the other path
+ dbAdapter = new GenericDatabaseAdapter();
+ query = processor.getQuery(dbAdapter, "myTable", null, Arrays.asList("id", "DATE_CREATED", "TIME_CREATED"), "type = \"CUSTOMER\"", stateManager.getState(Scope.CLUSTER).toMap());
+ assertEquals("SELECT * FROM myTable WHERE id > 509 AND DATE_CREATED >= '2016-03-07 12:34:56' AND TIME_CREATED >= '12:34:57' AND (type = \"CUSTOMER\")", query);
}
@Test(expected = IllegalArgumentException.class)