http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryRecord.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryRecord.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryRecord.java new file mode 100644 index 0000000..5982908 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/QueryRecord.java @@ -0,0 +1,587 @@ +/* + * 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 java.io.BufferedInputStream; +import java.io.Closeable; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; + +import org.apache.calcite.config.CalciteConnectionProperty; +import org.apache.calcite.config.Lex; +import org.apache.calcite.jdbc.CalciteConnection; +import org.apache.calcite.schema.SchemaPlus; +import org.apache.calcite.sql.parser.SqlParser; +import org.apache.nifi.annotation.behavior.DynamicProperty; +import org.apache.nifi.annotation.behavior.DynamicRelationship; +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.SideEffectFree; +import org.apache.nifi.annotation.behavior.SupportsBatching; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.components.Validator; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.flowfile.attributes.CoreAttributes; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.ProcessorInitializationContext; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.OutputStreamCallback; +import org.apache.nifi.queryrecord.FlowFileTable; +import org.apache.nifi.serialization.RecordSetWriter; +import org.apache.nifi.serialization.RecordSetWriterFactory; +import org.apache.nifi.serialization.RecordReaderFactory; +import org.apache.nifi.serialization.WriteResult; +import org.apache.nifi.serialization.record.ResultSetRecordSet; +import org.apache.nifi.util.StopWatch; + +@EventDriven +@SideEffectFree +@SupportsBatching +@Tags({"sql", "query", "calcite", "route", "record", "transform", "select", "update", "modify", "etl", "filter", "record", "csv", "json", "logs", "text", "avro", "aggregate"}) +@InputRequirement(Requirement.INPUT_REQUIRED) +@CapabilityDescription("Evaluates one or more SQL queries against the contents of a FlowFile. The result of the " + + "SQL query then becomes the content of the output FlowFile. This can be used, for example, " + + "for field-specific filtering, transformation, and row-level filtering. " + + "Columns can be renamed, simple calculations and aggregations performed, etc. " + + "The Processor is configured with a Record Reader Controller Service and a Record Writer service so as to allow flexibility in incoming and outgoing data formats. " + + "The Processor must be configured with at least one user-defined property. The name of the Property " + + "is the Relationship to route data to, and the value of the Property is a SQL SELECT statement that is used to specify how input data should be transformed/filtered. " + + "The SQL statement must be valid ANSI SQL and is powered by Apache Calcite. " + + "If the transformation fails, the original FlowFile is routed to the 'failure' relationship. Otherwise, the data selected will be routed to the associated " + + "relationship. See the Processor Usage documentation for more information.") +@DynamicRelationship(name="<Property Name>", description="Each user-defined property defines a new Relationship for this Processor.") +@DynamicProperty(name = "The name of the relationship to route data to", value="A SQL SELECT statement that is used to determine what data should be routed to this " + + "relationship.", supportsExpressionLanguage=true, description="Each user-defined property specifies a SQL SELECT statement to run over the data, with the data " + + "that is selected being routed to the relationship whose name is the property name") +public class QueryRecord extends AbstractProcessor { + static final PropertyDescriptor RECORD_READER_FACTORY = new PropertyDescriptor.Builder() + .name("record-reader") + .displayName("Record Reader") + .description("Specifies the Controller Service to use for parsing incoming data and determining the data's schema") + .identifiesControllerService(RecordReaderFactory.class) + .required(true) + .build(); + static final PropertyDescriptor RECORD_WRITER_FACTORY = new PropertyDescriptor.Builder() + .name("record-writer") + .displayName("Record Writer") + .description("Specifies the Controller Service to use for writing results to a FlowFile") + .identifiesControllerService(RecordSetWriterFactory.class) + .required(true) + .build(); + static final PropertyDescriptor INCLUDE_ZERO_RECORD_FLOWFILES = new PropertyDescriptor.Builder() + .name("include-zero-record-flowfiles") + .displayName("Include Zero Record FlowFiles") + .description("When running the SQL statement against an incoming FlowFile, if the result has no data, " + + "this property specifies whether or not a FlowFile will be sent to the corresponding relationship") + .expressionLanguageSupported(false) + .allowableValues("true", "false") + .defaultValue("true") + .required(true) + .build(); + static final PropertyDescriptor CACHE_SCHEMA = new PropertyDescriptor.Builder() + .name("cache-schema") + .displayName("Cache Schema") + .description("Parsing the SQL query and deriving the FlowFile's schema is relatively expensive. If this value is set to true, " + + "the Processor will cache these values so that the Processor is much more efficient and much faster. However, if this is done, " + + "then the schema that is derived for the first FlowFile processed must apply to all FlowFiles. If all FlowFiles will not have the exact " + + "same schema, or if the SQL SELECT statement uses the Expression Language, this value should be set to false.") + .expressionLanguageSupported(false) + .allowableValues("true", "false") + .defaultValue("true") + .required(true) + .build(); + + public static final Relationship REL_ORIGINAL = new Relationship.Builder() + .name("original") + .description("The original FlowFile is routed to this relationship") + .build(); + public static final Relationship REL_FAILURE = new Relationship.Builder() + .name("failure") + .description("If a FlowFile fails processing for any reason (for example, the SQL " + + "statement contains columns not present in input data), the original FlowFile it will " + + "be routed to this relationship") + .build(); + + private List<PropertyDescriptor> properties; + private final Set<Relationship> relationships = Collections.synchronizedSet(new HashSet<>()); + + private final Map<String, BlockingQueue<CachedStatement>> statementQueues = new HashMap<>(); + + @Override + protected void init(final ProcessorInitializationContext context) { + try { + DriverManager.registerDriver(new org.apache.calcite.jdbc.Driver()); + } catch (final SQLException e) { + throw new ProcessException("Failed to load Calcite JDBC Driver", e); + } + + final List<PropertyDescriptor> properties = new ArrayList<>(); + properties.add(RECORD_READER_FACTORY); + properties.add(RECORD_WRITER_FACTORY); + properties.add(INCLUDE_ZERO_RECORD_FLOWFILES); + properties.add(CACHE_SCHEMA); + this.properties = Collections.unmodifiableList(properties); + + relationships.add(REL_FAILURE); + relationships.add(REL_ORIGINAL); + } + + @Override + public Set<Relationship> getRelationships() { + return relationships; + } + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return properties; + } + + @Override + public void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue) { + if (!descriptor.isDynamic()) { + return; + } + + final Relationship relationship = new Relationship.Builder() + .name(descriptor.getName()) + .description("User-defined relationship that specifies where data that matches the specified SQL query should be routed") + .build(); + + if (newValue == null) { + relationships.remove(relationship); + } else { + relationships.add(relationship); + } + } + + @Override + protected Collection<ValidationResult> customValidate(final ValidationContext validationContext) { + final boolean cache = validationContext.getProperty(CACHE_SCHEMA).asBoolean(); + if (cache) { + for (final PropertyDescriptor descriptor : validationContext.getProperties().keySet()) { + if (descriptor.isDynamic() && validationContext.isExpressionLanguagePresent(validationContext.getProperty(descriptor).getValue())) { + return Collections.singleton(new ValidationResult.Builder() + .subject("Cache Schema") + .input("true") + .valid(false) + .explanation("Cannot have 'Cache Schema' property set to true if any SQL statement makes use of the Expression Language") + .build()); + } + } + } + + return Collections.emptyList(); + } + + @Override + protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) { + return new PropertyDescriptor.Builder() + .name(propertyDescriptorName) + .description("SQL select statement specifies how data should be filtered/transformed. " + + "SQL SELECT should select from the FLOWFILE table") + .required(false) + .dynamic(true) + .expressionLanguageSupported(true) + .addValidator(new SqlValidator()) + .build(); + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) { + final FlowFile original = session.get(); + if (original == null) { + return; + } + + final StopWatch stopWatch = new StopWatch(true); + + final RecordSetWriterFactory resultSetWriterFactory = context.getProperty(RECORD_WRITER_FACTORY) + .asControllerService(RecordSetWriterFactory.class); + final RecordReaderFactory recordParserFactory = context.getProperty(RECORD_READER_FACTORY) + .asControllerService(RecordReaderFactory.class); + + final Map<FlowFile, Relationship> transformedFlowFiles = new HashMap<>(); + final Set<FlowFile> createdFlowFiles = new HashSet<>(); + + int recordsRead = 0; + + try { + final RecordSetWriter resultSetWriter; + try (final InputStream rawIn = session.read(original); + final InputStream in = new BufferedInputStream(rawIn)) { + resultSetWriter = resultSetWriterFactory.createWriter(getLogger(), original, in); + } + + for (final PropertyDescriptor descriptor : context.getProperties().keySet()) { + if (!descriptor.isDynamic()) { + continue; + } + + final Relationship relationship = new Relationship.Builder().name(descriptor.getName()).build(); + + // We have to fork a child because we may need to read the input FlowFile more than once, + // and we cannot call session.read() on the original FlowFile while we are within a write + // callback for the original FlowFile. + FlowFile transformed = session.create(original); + boolean flowFileRemoved = false; + + try { + final String sql = context.getProperty(descriptor).evaluateAttributeExpressions(original).getValue(); + final AtomicReference<WriteResult> writeResultRef = new AtomicReference<>(); + final QueryResult queryResult; + if (context.getProperty(CACHE_SCHEMA).asBoolean()) { + queryResult = queryWithCache(session, original, sql, context, recordParserFactory); + } else { + queryResult = query(session, original, sql, context, recordParserFactory); + } + + try { + final ResultSet rs = queryResult.getResultSet(); + transformed = session.write(transformed, new OutputStreamCallback() { + @Override + public void process(final OutputStream out) throws IOException { + try { + final ResultSetRecordSet recordSet = new ResultSetRecordSet(rs); + writeResultRef.set(resultSetWriter.write(recordSet, out)); + } catch (final Exception e) { + throw new IOException(e); + } + } + }); + } finally { + closeQuietly(queryResult); + } + + recordsRead = Math.max(recordsRead, queryResult.getRecordsRead()); + final WriteResult result = writeResultRef.get(); + if (result.getRecordCount() == 0 && !context.getProperty(INCLUDE_ZERO_RECORD_FLOWFILES).asBoolean()) { + session.remove(transformed); + flowFileRemoved = true; + transformedFlowFiles.remove(transformed); + getLogger().info("Transformed {} but the result contained no data so will not pass on a FlowFile", new Object[] {original}); + } else { + final Map<String, String> attributesToAdd = new HashMap<>(); + if (result.getAttributes() != null) { + attributesToAdd.putAll(result.getAttributes()); + } + + attributesToAdd.put(CoreAttributes.MIME_TYPE.key(), resultSetWriter.getMimeType()); + attributesToAdd.put("record.count", String.valueOf(result.getRecordCount())); + transformed = session.putAllAttributes(transformed, attributesToAdd); + transformedFlowFiles.put(transformed, relationship); + + session.adjustCounter("Records Written", result.getRecordCount(), false); + } + } finally { + // Ensure that we have the FlowFile in the set in case we throw any Exception + if (!flowFileRemoved) { + createdFlowFiles.add(transformed); + } + } + } + + final long elapsedMillis = stopWatch.getElapsed(TimeUnit.MILLISECONDS); + if (transformedFlowFiles.size() > 0) { + session.getProvenanceReporter().fork(original, transformedFlowFiles.keySet(), elapsedMillis); + + for (final Map.Entry<FlowFile, Relationship> entry : transformedFlowFiles.entrySet()) { + final FlowFile transformed = entry.getKey(); + final Relationship relationship = entry.getValue(); + + session.getProvenanceReporter().route(transformed, relationship); + session.transfer(transformed, relationship); + } + } + + getLogger().info("Successfully queried {} in {} millis", new Object[] {original, elapsedMillis}); + session.transfer(original, REL_ORIGINAL); + } catch (final SQLException e) { + getLogger().error("Unable to query {} due to {}", new Object[] {original, e.getCause() == null ? e : e.getCause()}); + session.remove(createdFlowFiles); + session.transfer(original, REL_FAILURE); + } catch (final Exception e) { + getLogger().error("Unable to query {} due to {}", new Object[] {original, e}); + session.remove(createdFlowFiles); + session.transfer(original, REL_FAILURE); + } + + session.adjustCounter("Records Read", recordsRead, false); + } + + private synchronized CachedStatement getStatement(final String sql, final Supplier<CalciteConnection> connectionSupplier, final ProcessSession session, + final FlowFile flowFile, final RecordReaderFactory recordReaderFactory) throws SQLException { + + final BlockingQueue<CachedStatement> statementQueue = statementQueues.get(sql); + if (statementQueue == null) { + return buildCachedStatement(sql, connectionSupplier, session, flowFile, recordReaderFactory); + } + + final CachedStatement cachedStmt = statementQueue.poll(); + if (cachedStmt != null) { + return cachedStmt; + } + + return buildCachedStatement(sql, connectionSupplier, session, flowFile, recordReaderFactory); + } + + private CachedStatement buildCachedStatement(final String sql, final Supplier<CalciteConnection> connectionSupplier, final ProcessSession session, + final FlowFile flowFile, final RecordReaderFactory recordReaderFactory) throws SQLException { + + final CalciteConnection connection = connectionSupplier.get(); + final SchemaPlus rootSchema = connection.getRootSchema(); + + final FlowFileTable<?, ?> flowFileTable = new FlowFileTable<>(session, flowFile, recordReaderFactory, getLogger()); + rootSchema.add("FLOWFILE", flowFileTable); + rootSchema.setCacheEnabled(false); + + final PreparedStatement stmt = connection.prepareStatement(sql); + return new CachedStatement(stmt, flowFileTable, connection); + } + + @OnStopped + public synchronized void cleanup() { + for (final BlockingQueue<CachedStatement> statementQueue : statementQueues.values()) { + CachedStatement stmt; + while ((stmt = statementQueue.poll()) != null) { + closeQuietly(stmt.getStatement(), stmt.getConnection()); + } + } + + statementQueues.clear(); + } + + @OnScheduled + public synchronized void setupQueues(final ProcessContext context) { + // Create a Queue of PreparedStatements for each property that is user-defined. This allows us to easily poll the + // queue and add as necessary, knowing that the queue already exists. + for (final PropertyDescriptor descriptor : context.getProperties().keySet()) { + if (!descriptor.isDynamic()) { + continue; + } + + final String sql = context.getProperty(descriptor).evaluateAttributeExpressions().getValue(); + final BlockingQueue<CachedStatement> queue = new LinkedBlockingQueue<>(context.getMaxConcurrentTasks()); + statementQueues.put(sql, queue); + } + } + + protected QueryResult queryWithCache(final ProcessSession session, final FlowFile flowFile, final String sql, final ProcessContext context, + final RecordReaderFactory recordParserFactory) throws SQLException { + + final Supplier<CalciteConnection> connectionSupplier = () -> { + final Properties properties = new Properties(); + properties.put(CalciteConnectionProperty.LEX.camelName(), Lex.MYSQL_ANSI.name()); + + try { + final Connection connection = DriverManager.getConnection("jdbc:calcite:", properties); + final CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class); + return calciteConnection; + } catch (final Exception e) { + throw new ProcessException(e); + } + }; + + final CachedStatement cachedStatement = getStatement(sql, connectionSupplier, session, flowFile, recordParserFactory); + final PreparedStatement stmt = cachedStatement.getStatement(); + final FlowFileTable<?, ?> table = cachedStatement.getTable(); + table.setFlowFile(session, flowFile); + + final ResultSet rs = stmt.executeQuery(); + + return new QueryResult() { + @Override + public void close() throws IOException { + final BlockingQueue<CachedStatement> statementQueue = statementQueues.get(sql); + if (statementQueue == null || !statementQueue.offer(cachedStatement)) { + try { + cachedStatement.getConnection().close(); + } catch (SQLException e) { + throw new IOException("Failed to close statement", e); + } + } + } + + @Override + public ResultSet getResultSet() { + return rs; + } + + @Override + public int getRecordsRead() { + return table.getRecordsRead(); + } + + }; + } + + protected QueryResult query(final ProcessSession session, final FlowFile flowFile, final String sql, final ProcessContext context, + final RecordReaderFactory recordParserFactory) throws SQLException { + + final Properties properties = new Properties(); + properties.put(CalciteConnectionProperty.LEX.camelName(), Lex.JAVA.name()); + + Connection connection = null; + ResultSet resultSet = null; + Statement statement = null; + try { + connection = DriverManager.getConnection("jdbc:calcite:", properties); + final CalciteConnection calciteConnection = connection.unwrap(CalciteConnection.class); + final SchemaPlus rootSchema = calciteConnection.getRootSchema(); + + final FlowFileTable<?, ?> flowFileTable = new FlowFileTable<>(session, flowFile, recordParserFactory, getLogger()); + rootSchema.add("FLOWFILE", flowFileTable); + rootSchema.setCacheEnabled(false); + + statement = connection.createStatement(); + resultSet = statement.executeQuery(sql); + + final ResultSet rs = resultSet; + final Statement stmt = statement; + final Connection conn = connection; + return new QueryResult() { + @Override + public void close() throws IOException { + closeQuietly(rs, stmt, conn); + } + + @Override + public ResultSet getResultSet() { + return rs; + } + + @Override + public int getRecordsRead() { + return flowFileTable.getRecordsRead(); + } + }; + } catch (final Exception e) { + closeQuietly(resultSet, statement, connection); + throw e; + } + } + + private void closeQuietly(final AutoCloseable... closeables) { + if (closeables == null) { + return; + } + + for (final AutoCloseable closeable : closeables) { + if (closeable == null) { + continue; + } + + try { + closeable.close(); + } catch (final Exception e) { + getLogger().warn("Failed to close SQL resource", e); + } + } + } + + private static class SqlValidator implements Validator { + @Override + public ValidationResult validate(final String subject, final String input, final ValidationContext context) { + if (context.isExpressionLanguagePresent(input)) { + return new ValidationResult.Builder() + .input(input) + .subject(subject) + .valid(true) + .explanation("Expression Language Present") + .build(); + } + + final String substituted = context.newPropertyValue(input).evaluateAttributeExpressions().getValue(); + final SqlParser parser = SqlParser.create(substituted); + try { + parser.parseStmt(); + return new ValidationResult.Builder() + .subject(subject) + .input(input) + .valid(true) + .build(); + } catch (final Exception e) { + return new ValidationResult.Builder() + .subject(subject) + .input(input) + .valid(false) + .explanation("Not a valid SQL Statement: " + e.getMessage()) + .build(); + } + } + } + + private static interface QueryResult extends Closeable { + ResultSet getResultSet(); + + int getRecordsRead(); + } + + private static class CachedStatement { + private final FlowFileTable<?, ?> table; + private final PreparedStatement statement; + private final Connection connection; + + public CachedStatement(final PreparedStatement statement, final FlowFileTable<?, ?> table, final Connection connection) { + this.statement = statement; + this.table = table; + this.connection = connection; + } + + public FlowFileTable<?, ?> getTable() { + return table; + } + + public PreparedStatement getStatement() { + return statement; + } + + public Connection getConnection() { + return connection; + } + } +}
http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileEnumerator.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileEnumerator.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileEnumerator.java deleted file mode 100644 index 7daa002..0000000 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileEnumerator.java +++ /dev/null @@ -1,146 +0,0 @@ -/* - * 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.queryflowfile; - -import java.io.IOException; -import java.io.InputStream; - -import org.apache.calcite.linq4j.Enumerator; -import org.apache.nifi.flowfile.FlowFile; -import org.apache.nifi.logging.ComponentLog; -import org.apache.nifi.processor.ProcessSession; -import org.apache.nifi.processor.exception.ProcessException; -import org.apache.nifi.serialization.MalformedRecordException; -import org.apache.nifi.serialization.RecordReader; -import org.apache.nifi.serialization.RowRecordReaderFactory; -import org.apache.nifi.serialization.record.Record; - -public class FlowFileEnumerator<InternalType> implements Enumerator<Object> { - private final ProcessSession session; - private final FlowFile flowFile; - private final ComponentLog logger; - private final RowRecordReaderFactory recordParserFactory; - private final int[] fields; - - private InputStream rawIn; - private Object currentRow; - private RecordReader recordParser; - - public FlowFileEnumerator(final ProcessSession session, final FlowFile flowFile, final ComponentLog logger, final RowRecordReaderFactory parserFactory, final int[] fields) { - this.session = session; - this.flowFile = flowFile; - this.recordParserFactory = parserFactory; - this.logger = logger; - this.fields = fields; - reset(); - } - - @Override - public Object current() { - return currentRow; - } - - @Override - public boolean moveNext() { - currentRow = null; - while (currentRow == null) { - try { - currentRow = filterColumns(recordParser.nextRecord()); - break; - } catch (final IOException e) { - logger.error("Failed to read next record in stream for " + flowFile + ". Assuming end of stream.", e); - currentRow = null; - break; - } catch (final MalformedRecordException mre) { - logger.error("Failed to parse record in stream for " + flowFile + ". Will skip record and continue reading", mre); - } - } - - if (currentRow == null) { - // If we are out of data, close the InputStream. We do this because - // Calcite does not necessarily call our close() method. - close(); - } - return (currentRow != null); - } - - private Object filterColumns(final Record record) { - if (record == null) { - return null; - } - - final Object[] row = record.getValues(); - - // If we want no fields or if the row is null, just return null - if (fields == null || row == null) { - return row; - } - - // If we want only a single field, then Calcite is going to expect us to return - // the actual value, NOT a 1-element array of values. - if (fields.length == 1) { - final int desiredCellIndex = fields[0]; - return row[desiredCellIndex]; - } - - // Create a new Object array that contains only the desired fields. - final Object[] filtered = new Object[fields.length]; - for (int i = 0; i < fields.length; i++) { - final int indexToKeep = fields[i]; - filtered[i] = row[indexToKeep]; - } - - return filtered; - } - - @Override - public void reset() { - if (rawIn != null) { - try { - rawIn.close(); - } catch (final Exception e) { - logger.warn("Could not close FlowFile's input due to " + e, e); - } - } - - rawIn = session.read(flowFile); - - try { - recordParser = recordParserFactory.createRecordReader(flowFile, rawIn, logger); - } catch (final MalformedRecordException | IOException e) { - throw new ProcessException("Failed to reset stream", e); - } - } - - @Override - public void close() { - if (recordParser != null) { - try { - recordParser.close(); - } catch (final Exception e) { - logger.warn("Failed to close decorated source for " + flowFile, e); - } - } - - try { - rawIn.close(); - } catch (final Exception e) { - logger.warn("Failed to close InputStream for " + flowFile, e); - } - } -} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileProjectTableScanRule.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileProjectTableScanRule.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileProjectTableScanRule.java deleted file mode 100644 index c5179c9..0000000 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileProjectTableScanRule.java +++ /dev/null @@ -1,76 +0,0 @@ -/* - * 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.queryflowfile; - -import java.util.List; - -import org.apache.calcite.plan.RelOptRule; -import org.apache.calcite.plan.RelOptRuleCall; -import org.apache.calcite.rel.logical.LogicalProject; -import org.apache.calcite.rex.RexInputRef; -import org.apache.calcite.rex.RexNode; - -/** - * Planner rule that projects from a {@link FlowFileTableScan} scan just the columns - * needed to satisfy a projection. If the projection's expressions are trivial, - * the projection is removed. - */ -public class FlowFileProjectTableScanRule extends RelOptRule { - public static final FlowFileProjectTableScanRule INSTANCE = new FlowFileProjectTableScanRule(); - - private FlowFileProjectTableScanRule() { - super( - operand(LogicalProject.class, - operand(FlowFileTableScan.class, none())), - "FlowFileProjectTableScanRule"); - } - - @Override - public void onMatch(RelOptRuleCall call) { - final LogicalProject project = call.rel(0); - final FlowFileTableScan scan = call.rel(1); - final int[] fields = getProjectFields(project.getProjects()); - - if (fields == null) { - // Project contains expressions more complex than just field references. - return; - } - - call.transformTo( - new FlowFileTableScan( - scan.getCluster(), - scan.getTable(), - scan.flowFileTable, - fields)); - } - - private int[] getProjectFields(List<RexNode> exps) { - final int[] fields = new int[exps.size()]; - - for (int i = 0; i < exps.size(); i++) { - final RexNode exp = exps.get(i); - - if (exp instanceof RexInputRef) { - fields[i] = ((RexInputRef) exp).getIndex(); - } else { - return null; // not a simple projection - } - } - - return fields; - } -} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileTable.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileTable.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileTable.java deleted file mode 100644 index 27f0c42..0000000 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileTable.java +++ /dev/null @@ -1,203 +0,0 @@ -/* - * 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.queryflowfile; - -import java.io.IOException; -import java.io.InputStream; -import java.lang.reflect.Type; -import java.util.ArrayList; -import java.util.List; - -import org.apache.calcite.adapter.java.JavaTypeFactory; -import org.apache.calcite.linq4j.AbstractEnumerable; -import org.apache.calcite.linq4j.Enumerable; -import org.apache.calcite.linq4j.Enumerator; -import org.apache.calcite.linq4j.QueryProvider; -import org.apache.calcite.linq4j.Queryable; -import org.apache.calcite.linq4j.tree.Expression; -import org.apache.calcite.plan.RelOptTable; -import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.type.RelDataType; -import org.apache.calcite.rel.type.RelDataTypeFactory; -import org.apache.calcite.schema.QueryableTable; -import org.apache.calcite.schema.Schema.TableType; -import org.apache.calcite.schema.SchemaPlus; -import org.apache.calcite.schema.Schemas; -import org.apache.calcite.schema.TranslatableTable; -import org.apache.calcite.schema.impl.AbstractTable; -import org.apache.calcite.util.Pair; -import org.apache.nifi.flowfile.FlowFile; -import org.apache.nifi.logging.ComponentLog; -import org.apache.nifi.processor.ProcessSession; -import org.apache.nifi.processor.exception.ProcessException; -import org.apache.nifi.serialization.MalformedRecordException; -import org.apache.nifi.serialization.RecordReader; -import org.apache.nifi.serialization.RowRecordReaderFactory; -import org.apache.nifi.serialization.record.DataType; -import org.apache.nifi.serialization.record.RecordField; -import org.apache.nifi.serialization.record.RecordSchema; - - -public class FlowFileTable<S, E> extends AbstractTable implements QueryableTable, TranslatableTable { - - private final RowRecordReaderFactory recordParserFactory; - private final ComponentLog logger; - - private RecordSchema recordSchema; - private RelDataType relDataType = null; - - private volatile ProcessSession session; - private volatile FlowFile flowFile; - - /** - * Creates a FlowFile table. - */ - public FlowFileTable(final ProcessSession session, final FlowFile flowFile, final RowRecordReaderFactory recordParserFactory, final ComponentLog logger) { - this.session = session; - this.flowFile = flowFile; - this.recordParserFactory = recordParserFactory; - this.logger = logger; - } - - public void setFlowFile(final ProcessSession session, final FlowFile flowFile) { - this.session = session; - this.flowFile = flowFile; - } - - - @Override - public String toString() { - return "FlowFileTable"; - } - - /** - * Returns an enumerable over a given projection of the fields. - * - * <p> - * Called from generated code. - */ - public Enumerable<Object> project(final int[] fields) { - return new AbstractEnumerable<Object>() { - @Override - @SuppressWarnings({"unchecked", "rawtypes"}) - public Enumerator<Object> enumerator() { - return new FlowFileEnumerator(session, flowFile, logger, recordParserFactory, fields); - } - }; - } - - @Override - @SuppressWarnings("rawtypes") - public Expression getExpression(final SchemaPlus schema, final String tableName, final Class clazz) { - return Schemas.tableExpression(schema, getElementType(), tableName, clazz); - } - - @Override - public Type getElementType() { - return Object[].class; - } - - @Override - public <T> Queryable<T> asQueryable(final QueryProvider queryProvider, final SchemaPlus schema, final String tableName) { - throw new UnsupportedOperationException(); - } - - @Override - public RelNode toRel(final RelOptTable.ToRelContext context, final RelOptTable relOptTable) { - // Request all fields. - final int fieldCount = relOptTable.getRowType().getFieldCount(); - final int[] fields = new int[fieldCount]; - for (int i = 0; i < fieldCount; i++) { - fields[i] = i; - } - - return new FlowFileTableScan(context.getCluster(), relOptTable, this, fields); - } - - @Override - public RelDataType getRowType(final RelDataTypeFactory typeFactory) { - if (relDataType != null) { - return relDataType; - } - - RecordSchema schema; - try (final InputStream in = session.read(flowFile)) { - final RecordReader recordParser = recordParserFactory.createRecordReader(flowFile, in, logger); - schema = recordParser.getSchema(); - } catch (final MalformedRecordException | IOException e) { - throw new ProcessException("Failed to determine schema of data records for " + flowFile, e); - } - - final List<String> names = new ArrayList<>(); - final List<RelDataType> types = new ArrayList<>(); - - final JavaTypeFactory javaTypeFactory = (JavaTypeFactory) typeFactory; - for (final RecordField field : schema.getFields()) { - names.add(field.getFieldName()); - types.add(getRelDataType(field.getDataType(), javaTypeFactory)); - } - - logger.debug("Found Schema: {}", new Object[] {schema}); - - if (recordSchema == null) { - recordSchema = schema; - } - - relDataType = typeFactory.createStructType(Pair.zip(names, types)); - return relDataType; - } - - private RelDataType getRelDataType(final DataType fieldType, final JavaTypeFactory typeFactory) { - switch (fieldType.getFieldType()) { - case BOOLEAN: - return typeFactory.createJavaType(boolean.class); - case BYTE: - return typeFactory.createJavaType(byte.class); - case CHAR: - return typeFactory.createJavaType(char.class); - case DATE: - return typeFactory.createJavaType(java.sql.Date.class); - case DOUBLE: - return typeFactory.createJavaType(double.class); - case FLOAT: - return typeFactory.createJavaType(float.class); - case INT: - return typeFactory.createJavaType(int.class); - case SHORT: - return typeFactory.createJavaType(short.class); - case TIME: - return typeFactory.createJavaType(java.sql.Time.class); - case TIMESTAMP: - return typeFactory.createJavaType(java.sql.Timestamp.class); - case LONG: - return typeFactory.createJavaType(long.class); - case STRING: - return typeFactory.createJavaType(String.class); - case ARRAY: - return typeFactory.createJavaType(Object[].class); - case RECORD: - return typeFactory.createJavaType(Object.class); - } - - throw new IllegalArgumentException("Unknown Record Field Type: " + fieldType); - } - - @Override - public TableType getJdbcTableType() { - return TableType.TEMPORARY_TABLE; - } -} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileTableScan.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileTableScan.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileTableScan.java deleted file mode 100644 index ad3a1c3..0000000 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryflowfile/FlowFileTableScan.java +++ /dev/null @@ -1,91 +0,0 @@ -/* - * 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.queryflowfile; - -import java.util.List; - -import org.apache.calcite.adapter.enumerable.EnumerableConvention; -import org.apache.calcite.adapter.enumerable.EnumerableRel; -import org.apache.calcite.adapter.enumerable.EnumerableRelImplementor; -import org.apache.calcite.adapter.enumerable.PhysType; -import org.apache.calcite.adapter.enumerable.PhysTypeImpl; -import org.apache.calcite.linq4j.tree.Blocks; -import org.apache.calcite.linq4j.tree.Expressions; -import org.apache.calcite.linq4j.tree.Primitive; -import org.apache.calcite.plan.RelOptCluster; -import org.apache.calcite.plan.RelOptPlanner; -import org.apache.calcite.plan.RelOptTable; -import org.apache.calcite.plan.RelTraitSet; -import org.apache.calcite.rel.RelNode; -import org.apache.calcite.rel.RelWriter; -import org.apache.calcite.rel.core.TableScan; -import org.apache.calcite.rel.type.RelDataType; -import org.apache.calcite.rel.type.RelDataTypeFactory; -import org.apache.calcite.rel.type.RelDataTypeField; - -/** - * Relational expression representing a scan of a FlowFile. - * - * <p> - * Like any table scan, it serves as a leaf node of a query tree. - * </p> - */ -public class FlowFileTableScan extends TableScan implements EnumerableRel { - final FlowFileTable<?, ?> flowFileTable; - final int[] fields; - - protected FlowFileTableScan(final RelOptCluster cluster, final RelOptTable table, final FlowFileTable<?, ?> flowFileTable, final int[] fields) { - super(cluster, cluster.traitSetOf(EnumerableConvention.INSTANCE), table); - - this.flowFileTable = flowFileTable; - this.fields = fields; - } - - @Override - public RelNode copy(final RelTraitSet traitSet, final List<RelNode> inputs) { - return new FlowFileTableScan(getCluster(), table, flowFileTable, fields); - } - - @Override - public RelWriter explainTerms(final RelWriter pw) { - return super.explainTerms(pw).item("fields", Primitive.asList(fields)); - } - - @Override - public RelDataType deriveRowType() { - final List<RelDataTypeField> fieldList = table.getRowType().getFieldList(); - final RelDataTypeFactory.FieldInfoBuilder builder = getCluster().getTypeFactory().builder(); - for (int field : fields) { - builder.add(fieldList.get(field)); - } - return builder.build(); - } - - @Override - public void register(RelOptPlanner planner) { - planner.addRule(FlowFileProjectTableScanRule.INSTANCE); - } - - @Override - public Result implement(EnumerableRelImplementor implementor, Prefer pref) { - PhysType physType = PhysTypeImpl.of(implementor.getTypeFactory(), getRowType(), pref.preferArray()); - - return implementor.result(physType, Blocks.toBlock( - Expressions.call(table.getExpression(FlowFileTable.class), "project", Expressions.constant(fields)))); - } - -} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileEnumerator.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileEnumerator.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileEnumerator.java new file mode 100644 index 0000000..e7b2c26 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileEnumerator.java @@ -0,0 +1,157 @@ +/* + * 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.queryrecord; + +import java.io.InputStream; + +import org.apache.calcite.linq4j.Enumerator; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.serialization.RecordReader; +import org.apache.nifi.serialization.RecordReaderFactory; +import org.apache.nifi.serialization.record.Record; + +public class FlowFileEnumerator<InternalType> implements Enumerator<Object> { + private final ProcessSession session; + private final FlowFile flowFile; + private final ComponentLog logger; + private final RecordReaderFactory recordParserFactory; + private final int[] fields; + + private InputStream rawIn; + private Object currentRow; + private RecordReader recordParser; + private int recordsRead = 0; + + public FlowFileEnumerator(final ProcessSession session, final FlowFile flowFile, final ComponentLog logger, final RecordReaderFactory parserFactory, final int[] fields) { + this.session = session; + this.flowFile = flowFile; + this.recordParserFactory = parserFactory; + this.logger = logger; + this.fields = fields; + reset(); + } + + @Override + public Object current() { + return currentRow; + } + + @Override + public boolean moveNext() { + currentRow = null; + while (currentRow == null) { + try { + currentRow = filterColumns(recordParser.nextRecord()); + break; + } catch (final Exception e) { + throw new ProcessException("Failed to read next record in stream for " + flowFile, e); + } + } + + if (currentRow == null) { + // If we are out of data, close the InputStream. We do this because + // Calcite does not necessarily call our close() method. + close(); + try { + onFinish(); + } catch (final Exception e) { + logger.error("Failed to perform tasks when enumerator was finished", e); + } + + return false; + } + + recordsRead++; + return true; + } + + protected int getRecordsRead() { + return recordsRead; + } + + protected void onFinish() { + } + + private Object filterColumns(final Record record) { + if (record == null) { + return null; + } + + final Object[] row = record.getValues(); + + // If we want no fields or if the row is null, just return null + if (fields == null || row == null) { + return row; + } + + // If we want only a single field, then Calcite is going to expect us to return + // the actual value, NOT a 1-element array of values. + if (fields.length == 1) { + final int desiredCellIndex = fields[0]; + return row[desiredCellIndex]; + } + + // Create a new Object array that contains only the desired fields. + final Object[] filtered = new Object[fields.length]; + for (int i = 0; i < fields.length; i++) { + final int indexToKeep = fields[i]; + filtered[i] = row[indexToKeep]; + } + + return filtered; + } + + @Override + public void reset() { + if (rawIn != null) { + try { + rawIn.close(); + } catch (final Exception e) { + logger.warn("Could not close FlowFile's input due to " + e, e); + } + } + + rawIn = session.read(flowFile); + + try { + recordParser = recordParserFactory.createRecordReader(flowFile, rawIn, logger); + } catch (final Exception e) { + throw new ProcessException("Failed to reset stream", e); + } + } + + @Override + public void close() { + if (recordParser != null) { + try { + recordParser.close(); + } catch (final Exception e) { + logger.warn("Failed to close decorated source for " + flowFile, e); + } + } + + try { + rawIn.close(); + } catch (final Exception e) { + logger.warn("Failed to close InputStream for " + flowFile, e); + } + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileProjectTableScanRule.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileProjectTableScanRule.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileProjectTableScanRule.java new file mode 100644 index 0000000..a44a5d8 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileProjectTableScanRule.java @@ -0,0 +1,76 @@ +/* + * 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.queryrecord; + +import java.util.List; + +import org.apache.calcite.plan.RelOptRule; +import org.apache.calcite.plan.RelOptRuleCall; +import org.apache.calcite.rel.logical.LogicalProject; +import org.apache.calcite.rex.RexInputRef; +import org.apache.calcite.rex.RexNode; + +/** + * Planner rule that projects from a {@link FlowFileTableScan} scan just the columns + * needed to satisfy a projection. If the projection's expressions are trivial, + * the projection is removed. + */ +public class FlowFileProjectTableScanRule extends RelOptRule { + public static final FlowFileProjectTableScanRule INSTANCE = new FlowFileProjectTableScanRule(); + + private FlowFileProjectTableScanRule() { + super( + operand(LogicalProject.class, + operand(FlowFileTableScan.class, none())), + "FlowFileProjectTableScanRule"); + } + + @Override + public void onMatch(RelOptRuleCall call) { + final LogicalProject project = call.rel(0); + final FlowFileTableScan scan = call.rel(1); + final int[] fields = getProjectFields(project.getProjects()); + + if (fields == null) { + // Project contains expressions more complex than just field references. + return; + } + + call.transformTo( + new FlowFileTableScan( + scan.getCluster(), + scan.getTable(), + scan.flowFileTable, + fields)); + } + + private int[] getProjectFields(List<RexNode> exps) { + final int[] fields = new int[exps.size()]; + + for (int i = 0; i < exps.size(); i++) { + final RexNode exp = exps.get(i); + + if (exp instanceof RexInputRef) { + fields[i] = ((RexInputRef) exp).getIndex(); + } else { + return null; // not a simple projection + } + } + + return fields; + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileTable.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileTable.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileTable.java new file mode 100644 index 0000000..bd15dc2 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileTable.java @@ -0,0 +1,218 @@ +/* + * 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.queryrecord; + +import java.io.InputStream; +import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; + +import org.apache.calcite.adapter.java.JavaTypeFactory; +import org.apache.calcite.linq4j.AbstractEnumerable; +import org.apache.calcite.linq4j.Enumerable; +import org.apache.calcite.linq4j.Enumerator; +import org.apache.calcite.linq4j.QueryProvider; +import org.apache.calcite.linq4j.Queryable; +import org.apache.calcite.linq4j.tree.Expression; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.schema.QueryableTable; +import org.apache.calcite.schema.Schema.TableType; +import org.apache.calcite.schema.SchemaPlus; +import org.apache.calcite.schema.Schemas; +import org.apache.calcite.schema.TranslatableTable; +import org.apache.calcite.schema.impl.AbstractTable; +import org.apache.calcite.util.Pair; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.serialization.RecordReader; +import org.apache.nifi.serialization.RecordReaderFactory; +import org.apache.nifi.serialization.record.DataType; +import org.apache.nifi.serialization.record.RecordField; +import org.apache.nifi.serialization.record.RecordSchema; + + +public class FlowFileTable<S, E> extends AbstractTable implements QueryableTable, TranslatableTable { + + private final RecordReaderFactory recordParserFactory; + private final ComponentLog logger; + + private RecordSchema recordSchema; + private RelDataType relDataType = null; + + private volatile ProcessSession session; + private volatile FlowFile flowFile; + private volatile int maxRecordsRead; + + /** + * Creates a FlowFile table. + */ + public FlowFileTable(final ProcessSession session, final FlowFile flowFile, final RecordReaderFactory recordParserFactory, final ComponentLog logger) { + this.session = session; + this.flowFile = flowFile; + this.recordParserFactory = recordParserFactory; + this.logger = logger; + } + + public void setFlowFile(final ProcessSession session, final FlowFile flowFile) { + this.session = session; + this.flowFile = flowFile; + this.maxRecordsRead = 0; + } + + + @Override + public String toString() { + return "FlowFileTable"; + } + + /** + * Returns an enumerable over a given projection of the fields. + * + * <p> + * Called from generated code. + */ + public Enumerable<Object> project(final int[] fields) { + return new AbstractEnumerable<Object>() { + @Override + @SuppressWarnings({"unchecked", "rawtypes"}) + public Enumerator<Object> enumerator() { + return new FlowFileEnumerator(session, flowFile, logger, recordParserFactory, fields) { + @Override + protected void onFinish() { + final int recordCount = getRecordsRead(); + if (recordCount > maxRecordsRead) { + maxRecordsRead = recordCount; + } + } + }; + } + }; + } + + public int getRecordsRead() { + return maxRecordsRead; + } + + @Override + @SuppressWarnings("rawtypes") + public Expression getExpression(final SchemaPlus schema, final String tableName, final Class clazz) { + return Schemas.tableExpression(schema, getElementType(), tableName, clazz); + } + + @Override + public Type getElementType() { + return Object[].class; + } + + @Override + public <T> Queryable<T> asQueryable(final QueryProvider queryProvider, final SchemaPlus schema, final String tableName) { + throw new UnsupportedOperationException(); + } + + @Override + public RelNode toRel(final RelOptTable.ToRelContext context, final RelOptTable relOptTable) { + // Request all fields. + final int fieldCount = relOptTable.getRowType().getFieldCount(); + final int[] fields = new int[fieldCount]; + for (int i = 0; i < fieldCount; i++) { + fields[i] = i; + } + + return new FlowFileTableScan(context.getCluster(), relOptTable, this, fields); + } + + @Override + public RelDataType getRowType(final RelDataTypeFactory typeFactory) { + if (relDataType != null) { + return relDataType; + } + + RecordSchema schema; + try (final InputStream in = session.read(flowFile)) { + final RecordReader recordParser = recordParserFactory.createRecordReader(flowFile, in, logger); + schema = recordParser.getSchema(); + } catch (final Exception e) { + throw new ProcessException("Failed to determine schema of data records for " + flowFile, e); + } + + final List<String> names = new ArrayList<>(); + final List<RelDataType> types = new ArrayList<>(); + + final JavaTypeFactory javaTypeFactory = (JavaTypeFactory) typeFactory; + for (final RecordField field : schema.getFields()) { + names.add(field.getFieldName()); + types.add(getRelDataType(field.getDataType(), javaTypeFactory)); + } + + logger.debug("Found Schema: {}", new Object[] {schema}); + + if (recordSchema == null) { + recordSchema = schema; + } + + relDataType = typeFactory.createStructType(Pair.zip(names, types)); + return relDataType; + } + + private RelDataType getRelDataType(final DataType fieldType, final JavaTypeFactory typeFactory) { + switch (fieldType.getFieldType()) { + case BOOLEAN: + return typeFactory.createJavaType(boolean.class); + case BYTE: + return typeFactory.createJavaType(byte.class); + case CHAR: + return typeFactory.createJavaType(char.class); + case DATE: + return typeFactory.createJavaType(java.sql.Date.class); + case DOUBLE: + return typeFactory.createJavaType(double.class); + case FLOAT: + return typeFactory.createJavaType(float.class); + case INT: + return typeFactory.createJavaType(int.class); + case SHORT: + return typeFactory.createJavaType(short.class); + case TIME: + return typeFactory.createJavaType(java.sql.Time.class); + case TIMESTAMP: + return typeFactory.createJavaType(java.sql.Timestamp.class); + case LONG: + return typeFactory.createJavaType(long.class); + case STRING: + return typeFactory.createJavaType(String.class); + case ARRAY: + return typeFactory.createJavaType(Object[].class); + case RECORD: + return typeFactory.createJavaType(Object.class); + case MAP: + return typeFactory.createJavaType(HashMap.class); + } + + throw new IllegalArgumentException("Unknown Record Field Type: " + fieldType); + } + + @Override + public TableType getJdbcTableType() { + return TableType.TEMPORARY_TABLE; + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileTableScan.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileTableScan.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileTableScan.java new file mode 100644 index 0000000..afca202 --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/queryrecord/FlowFileTableScan.java @@ -0,0 +1,91 @@ +/* + * 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.queryrecord; + +import java.util.List; + +import org.apache.calcite.adapter.enumerable.EnumerableConvention; +import org.apache.calcite.adapter.enumerable.EnumerableRel; +import org.apache.calcite.adapter.enumerable.EnumerableRelImplementor; +import org.apache.calcite.adapter.enumerable.PhysType; +import org.apache.calcite.adapter.enumerable.PhysTypeImpl; +import org.apache.calcite.linq4j.tree.Blocks; +import org.apache.calcite.linq4j.tree.Expressions; +import org.apache.calcite.linq4j.tree.Primitive; +import org.apache.calcite.plan.RelOptCluster; +import org.apache.calcite.plan.RelOptPlanner; +import org.apache.calcite.plan.RelOptTable; +import org.apache.calcite.plan.RelTraitSet; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.RelWriter; +import org.apache.calcite.rel.core.TableScan; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.rel.type.RelDataTypeFactory; +import org.apache.calcite.rel.type.RelDataTypeField; + +/** + * Relational expression representing a scan of a FlowFile. + * + * <p> + * Like any table scan, it serves as a leaf node of a query tree. + * </p> + */ +public class FlowFileTableScan extends TableScan implements EnumerableRel { + final FlowFileTable<?, ?> flowFileTable; + final int[] fields; + + protected FlowFileTableScan(final RelOptCluster cluster, final RelOptTable table, final FlowFileTable<?, ?> flowFileTable, final int[] fields) { + super(cluster, cluster.traitSetOf(EnumerableConvention.INSTANCE), table); + + this.flowFileTable = flowFileTable; + this.fields = fields; + } + + @Override + public RelNode copy(final RelTraitSet traitSet, final List<RelNode> inputs) { + return new FlowFileTableScan(getCluster(), table, flowFileTable, fields); + } + + @Override + public RelWriter explainTerms(final RelWriter pw) { + return super.explainTerms(pw).item("fields", Primitive.asList(fields)); + } + + @Override + public RelDataType deriveRowType() { + final List<RelDataTypeField> fieldList = table.getRowType().getFieldList(); + final RelDataTypeFactory.FieldInfoBuilder builder = getCluster().getTypeFactory().builder(); + for (int field : fields) { + builder.add(fieldList.get(field)); + } + return builder.build(); + } + + @Override + public void register(RelOptPlanner planner) { + planner.addRule(FlowFileProjectTableScanRule.INSTANCE); + } + + @Override + public Result implement(EnumerableRelImplementor implementor, Prefer pref) { + PhysType physType = PhysTypeImpl.of(implementor.getTypeFactory(), getRowType(), pref.preferArray()); + + return implementor.result(physType, Blocks.toBlock( + Expressions.call(table.getExpression(FlowFileTable.class), "project", Expressions.constant(fields)))); + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor index 3891ee6..d85e663 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor @@ -76,7 +76,7 @@ org.apache.nifi.processors.standard.PutSyslog org.apache.nifi.processors.standard.PutTCP org.apache.nifi.processors.standard.PutUDP org.apache.nifi.processors.standard.QueryDatabaseTable -org.apache.nifi.processors.standard.QueryFlowFile +org.apache.nifi.processors.standard.QueryRecord org.apache.nifi.processors.standard.ReplaceText org.apache.nifi.processors.standard.RouteText org.apache.nifi.processors.standard.ReplaceTextWithMapping http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.QueryFlowFile/additionalDetails.html ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.QueryFlowFile/additionalDetails.html b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.QueryFlowFile/additionalDetails.html deleted file mode 100644 index 0dffc0d..0000000 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.QueryFlowFile/additionalDetails.html +++ /dev/null @@ -1,48 +0,0 @@ -<!DOCTYPE html> -<html lang="en"> - <!-- - 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. - --> - <head> - <meta charset="utf-8" /> - <title>QueryFlowFile</title> - - <link rel="stylesheet" href="../../css/component-usage.css" type="text/css" /> - </head> - - <body> - <p> - QueryFlowFile provides users a tremendous amount of power by leveraging an extremely well-known - syntax (SQL) to route, filter, transform, and query data as it traverses the system. In order to - provide the Processor with the maximum amount of flexibility, it is configured with a Controller - Service that is responsible for reading and parsing the incoming FlowFiles and a Controller Service - that is responsible for writing the results out. By using this paradigm, users are not forced to - convert their data from one format to another just to query it, and then transform the data back - into the form that they want. Rather, the appropriate Controller Service can easily be configured - and put to use for the appropriate data format. - </p> - - <p> - Rather than providing a single "SQL SELECT Statement" type of Property, this Processor makes use - of user-defined properties. Each user-defined property that is added to the Processor has a name - that becomes a new Relationship for the Processor and a corresponding SQL query that will be evaluated - against each FlowFile. This allows multiple SQL queries to be run against each FlowFile. - </p> - - <p> - The SQL syntax that is supported by this Processor is ANSI SQL and is powered by Apache Calcite. Please - note that identifiers are quoted using double-quotes, and column names/labels are case-insensitive. - </p> - </body> -</html> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/50ea1083/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.QueryRecord/additionalDetails.html ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.QueryRecord/additionalDetails.html b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.QueryRecord/additionalDetails.html new file mode 100644 index 0000000..93bbe2a --- /dev/null +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.QueryRecord/additionalDetails.html @@ -0,0 +1,48 @@ +<!DOCTYPE html> +<html lang="en"> + <!-- + 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. + --> + <head> + <meta charset="utf-8" /> + <title>QueryRecord</title> + + <link rel="stylesheet" href="../../css/component-usage.css" type="text/css" /> + </head> + + <body> + <p> + QueryRecord provides users a tremendous amount of power by leveraging an extremely well-known + syntax (SQL) to route, filter, transform, and query data as it traverses the system. In order to + provide the Processor with the maximum amount of flexibility, it is configured with a Controller + Service that is responsible for reading and parsing the incoming FlowFiles and a Controller Service + that is responsible for writing the results out. By using this paradigm, users are not forced to + convert their data from one format to another just to query it, and then transform the data back + into the form that they want. Rather, the appropriate Controller Service can easily be configured + and put to use for the appropriate data format. + </p> + + <p> + Rather than providing a single "SQL SELECT Statement" type of Property, this Processor makes use + of user-defined properties. Each user-defined property that is added to the Processor has a name + that becomes a new Relationship for the Processor and a corresponding SQL query that will be evaluated + against each FlowFile. This allows multiple SQL queries to be run against each FlowFile. + </p> + + <p> + The SQL syntax that is supported by this Processor is ANSI SQL and is powered by Apache Calcite. Please + note that identifiers are quoted using double-quotes, and column names/labels are case-insensitive. + </p> + </body> +</html> \ No newline at end of file
