diff --git a/.gitignore b/.gitignore index e299e15e43c4..fd80f9edf700 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ target bin +dependency-reduced-pom.xml .project .settings .classpath diff --git a/nifi-assembly/NOTICE b/nifi-assembly/NOTICE index 3c66982fdc92..455f47323cbd 100644 --- a/nifi-assembly/NOTICE +++ b/nifi-assembly/NOTICE @@ -1997,6 +1997,11 @@ The following binary components are provided under the Apache Software License v the terms of a BSD style license. The original software and related information is available at http://www.jcraft.com/jsch/. + + (ASLv2) DataStax Java Driver for Apache Cassandra - Core + The following NOTICE information applies: + DataStax Java Driver for Apache Cassandra - Core + Copyright (C) 2012-2017 DataStax Inc. (ASLv2) bytebuffer-collections The following NOTICE information applies: bytebuffer-collections diff --git a/nifi-assembly/pom.xml b/nifi-assembly/pom.xml index e3cdf316ca57..b69e8b698c54 100644 --- a/nifi-assembly/pom.xml +++ b/nifi-assembly/pom.xml @@ -1202,5 +1202,44 @@ language governing permissions and limitations under the License. --> false + + include-cql + + + false + + allProfiles + + + + + org.apache.nifi + nifi-cql-nar + 2.12.0-SNAPSHOT + nar + + + org.apache.nifi + nifi-cql-services-api-nar + 2.12.0-SNAPSHOT + nar + + + org.apache.nifi + nifi-cassandra-session-provider-service-nar + 2.12.0-SNAPSHOT + nar + + + org.apache.nifi + nifi-scylladb-session-provider-service-nar + 2.12.0-SNAPSHOT + nar + + + diff --git a/nifi-code-coverage/pom.xml b/nifi-code-coverage/pom.xml index 1be8e671e662..ab7cd13d85e4 100644 --- a/nifi-code-coverage/pom.xml +++ b/nifi-code-coverage/pom.xml @@ -695,6 +695,21 @@ nifi-box-services-api 2.12.0-SNAPSHOT + + org.apache.nifi + nifi-cql-processors + 2.12.0-SNAPSHOT + + + org.apache.nifi + nifi-cassandra-session-provider-service + 2.12.0-SNAPSHOT + + + org.apache.nifi + nifi-cql-services-api + 2.12.0-SNAPSHOT + org.apache.nifi nifi-cdc-api diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service-nar/pom.xml b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service-nar/pom.xml new file mode 100644 index 000000000000..96897195adb2 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service-nar/pom.xml @@ -0,0 +1,51 @@ + + + + + nifi-cql-bundle + org.apache.nifi + 2.12.0-SNAPSHOT + + 4.0.0 + + nifi-cassandra-session-provider-service-nar + nar + + + + + + com.google.guava + guava + provided + + + + + + + org.apache.nifi + nifi-cql-services-api-nar + 2.12.0-SNAPSHOT + nar + + + org.apache.nifi + nifi-cassandra-session-provider-service + 2.12.0-SNAPSHOT + + + \ No newline at end of file diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service-nar/src/main/resources/META-INF/LICENSE b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service-nar/src/main/resources/META-INF/LICENSE new file mode 100644 index 000000000000..c567ce25dfd7 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service-nar/src/main/resources/META-INF/LICENSE @@ -0,0 +1,352 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed 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. + +APACHE NIFI SUBCOMPONENTS: + +The Apache NiFi project contains subcomponents with separate copyright +notices and license terms. Your use of the source code for the these +subcomponents is subject to the terms and conditions of the following +licenses. + +This product bundles 'libffi' which is available under an MIT style license. + libffi - Copyright (c) 1996-2014 Anthony Green, Red Hat, Inc and others. + see https://github.com/java-native-access/jna/blob/master/native/libffi/LICENSE + + Permission is hereby granted, free of charge, to any person obtaining + a copy of this software and associated documentation files (the + ``Software''), to deal in the Software without restriction, including + without limitation the rights to use, copy, modify, merge, publish, + distribute, sublicense, and/or sell copies of the Software, and to + permit persons to whom the Software is furnished to do so, subject to + the following conditions: + + The above copyright notice and this permission notice shall be + included in all copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED ``AS IS'', WITHOUT WARRANTY OF ANY KIND, + EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF + MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. + IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY + CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, + TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE + SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + +This product bundles 'asm' which is available under a 3-Clause BSD style license. +For details see http://asm.ow2.org/asmdex-license.html + + Copyright (c) 2012 France Télécom + All rights reserved. + + Redistribution and use in source and binary forms, with or without + modification, are permitted provided that the following conditions + are met: + 1. Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. + 2. Redistributions in binary form must reproduce the above copyright + notice, this list of conditions and the following disclaimer in the + documentation and/or other materials provided with the distribution. + 3. Neither the name of the copyright holders nor the names of its + contributors may be used to endorse or promote products derived from + this software without specific prior written permission. + + THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" + AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE + LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR + CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF + SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN + CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) + ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF + THE POSSIBILITY OF SUCH DAMAGE. + + The binary distribution of this product bundles 'Bouncy Castle JDK 1.5' + under an MIT style license. + + Copyright (c) 2000 - 2015 The Legion of the Bouncy Castle Inc. (http://www.bouncycastle.org) + + Permission is hereby granted, free of charge, to any person obtaining a copy + of this software and associated documentation files (the "Software"), to deal + in the Software without restriction, including without limitation the rights + to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the Software is + furnished to do so, subject to the following conditions: + + The above copyright notice and this permission notice shall be included in + all copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + THE SOFTWARE. + +The binary distribution of this product bundles 'JNR x86asm' under an MIT +style license. + + Copyright (C) 2010 Wayne Meissner + Copyright (c) 2008-2009, Petr Kobalicek + + Permission is hereby granted, free of charge, to any person + obtaining a copy of this software and associated documentation + files (the "Software"), to deal in the Software without + restriction, including without limitation the rights to use, + copy, modify, merge, publish, distribute, sublicense, and/or sell + copies of the Software, and to permit persons to whom the + Software is furnished to do so, subject to the following + conditions: + + The above copyright notice and this permission notice shall be + included in all copies or substantial portions of the Software. + + THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES + OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT + HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, + WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING + FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR + OTHER DEALINGS IN THE SOFTWARE. + +This product bundles 'logback' which is dual-licensed under the EPL v1.0 +and the LGPL 2.1. + + Logback: the reliable, generic, fast and flexible logging framework. + + Copyright (C) 1999-2017, QOS.ch. All rights reserved. + + This program and the accompanying materials are dual-licensed under + either the terms of the Eclipse Public License v1.0 as published by + the Eclipse Foundation or (per the licensee's choosing) under the + terms of the GNU Lesser General Public License version 2.1 as + published by the Free Software Foundation. + +The binary distribution of this product bundles 'ANTLR 3' which is available +under a "3-clause BSD" license. For details see http://www.antlr.org/license.html + + Copyright (c) 2012 Terence Parr and Sam Harwell + All rights reserved. + Redistribution and use in source and binary forms, with or without modification, are permitted + provided that the following conditions are met: + + Redistributions of source code must retain the above copyright notice, this list of + conditions and the following disclaimer. + Redistributions in binary form must reproduce the above copyright notice, this list of + conditions and the following disclaimer in the documentation and/or other materials + provided with the distribution. + + Neither the name of the author nor the names of its contributors may be used to endorse + or promote products derived from this software without specific prior written permission. + + THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY + EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF + MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL + THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, + PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS + INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, + STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF + THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service-nar/src/main/resources/META-INF/NOTICE b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service-nar/src/main/resources/META-INF/NOTICE new file mode 100644 index 000000000000..8e2e0dcfb324 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service-nar/src/main/resources/META-INF/NOTICE @@ -0,0 +1,313 @@ +nifi-cassandra-services-nar +Copyright 2016-2020 The Apache Software Foundation + +This product includes software developed at +The Apache Software Foundation (http://www.apache.org/). + +****************** +Apache Software License v2 +****************** + +The following binary components are provided under the Apache Software License v2 + + (ASLv2) DataStax Java Driver for Apache Cassandra - Core + The following NOTICE information applies: + DataStax Java Driver for Apache Cassandra - Core + Copyright (C) 2012-2017 DataStax Inc. + + (ASLv2) Jackson JSON processor + The following NOTICE information applies: + # Jackson JSON processor + + Jackson is a high-performance, Free/Open Source JSON processing library. + It was originally written by Tatu Saloranta (tatu.saloranta@iki.fi), and has + been in development since 2007. + It is currently developed by a community of developers, as well as supported + commercially by FasterXML.com. + + ## Licensing + + Jackson core and extension components may licensed under different licenses. + To find the details that apply to this artifact see the accompanying LICENSE file. + For more information, including possible other licensing options, contact + FasterXML.com (http://fasterxml.com). + + ## Credits + + A list of contributors may be found from CREDITS file, which is included + in some artifacts (usually source distributions); but is always available + from the source code management (SCM) system project uses. + + (ASLv2) Apache Commons Codec + The following NOTICE information applies: + Apache Commons Codec + Copyright 2002-2014 The Apache Software Foundation + + src/test/org/apache/commons/codec/language/DoubleMetaphoneTest.java + contains test data from http://aspell.net/test/orig/batch0.tab. + Copyright (C) 2002 Kevin Atkinson (kevina@gnu.org) + + =============================================================================== + + The content of package org.apache.commons.codec.language.bm has been translated + from the original php source code available at http://stevemorse.org/phoneticinfo.htm + with permission from the original authors. + Original source copyright: + Copyright (c) 2008 Alexander Beider & Stephen P. Morse. + + (ASLv2) Apache Commons Lang + The following NOTICE information applies: + Apache Commons Lang + Copyright 2001-2017 The Apache Software Foundation + + This product includes software from the Spring Framework, + under the Apache License 2.0 (see: StringUtils.containsWhitespace()) + + (ASLv2) Guava + The following NOTICE information applies: + Guava + Copyright 2015 The Guava Authors + + (ASLv2) JSON-SMART + The following NOTICE information applies: + Copyright 2011 JSON-SMART authors + + (ASLv2) Dropwizard Metrics + The following NOTICE information applies: + Copyright (c) 2010-2013 Coda Hale, Yammer.com + + This product includes software developed by Coda Hale and Yammer, Inc. + + This product includes code derived from the JSR-166 project (ThreadLocalRandom, Striped64, + LongAdder), which was released with the following comments: + + Written by Doug Lea with assistance from members of JCP JSR-166 + Expert Group and released to the public domain, as explained at + http://creativecommons.org/publicdomain/zero/1.0/ + + (ASLv2) The Netty Project + The following NOTICE information applies: + Copyright 2014 The Netty Project + ------------------------------------------------------------------------------- + This product contains the extensions to Java Collections Framework which has + been derived from the works by JSR-166 EG, Doug Lea, and Jason T. Greene: + + * LICENSE: + * license/LICENSE.jsr166y.txt (Public Domain) + * HOMEPAGE: + * http://gee.cs.oswego.edu/cgi-bin/viewcvs.cgi/jsr166/ + * http://viewvc.jboss.org/cgi-bin/viewvc.cgi/jbosscache/experimental/jsr166/ + + This product contains a modified version of Robert Harder's Public Domain + Base64 Encoder and Decoder, which can be obtained at: + + * LICENSE: + * license/LICENSE.base64.txt (Public Domain) + * HOMEPAGE: + * http://iharder.sourceforge.net/current/java/base64/ + + This product contains a modified portion of 'Webbit', an event based + WebSocket and HTTP server, which can be obtained at: + + * LICENSE: + * license/LICENSE.webbit.txt (BSD License) + * HOMEPAGE: + * https://github.com/joewalnes/webbit + + This product contains a modified portion of 'SLF4J', a simple logging + facade for Java, which can be obtained at: + + * LICENSE: + * license/LICENSE.slf4j.txt (MIT License) + * HOMEPAGE: + * http://www.slf4j.org/ + + This product contains a modified portion of 'Apache Harmony', an open source + Java SE, which can be obtained at: + + * LICENSE: + * license/LICENSE.harmony.txt (Apache License 2.0) + * HOMEPAGE: + * http://archive.apache.org/dist/harmony/ + + This product contains a modified portion of 'jbzip2', a Java bzip2 compression + and decompression library written by Matthew J. Francis. It can be obtained at: + + * LICENSE: + * license/LICENSE.jbzip2.txt (MIT License) + * HOMEPAGE: + * https://code.google.com/p/jbzip2/ + + This product contains a modified portion of 'libdivsufsort', a C API library to construct + the suffix array and the Burrows-Wheeler transformed string for any input string of + a constant-size alphabet written by Yuta Mori. It can be obtained at: + + * LICENSE: + * license/LICENSE.libdivsufsort.txt (MIT License) + * HOMEPAGE: + * https://github.com/y-256/libdivsufsort + + This product contains a modified portion of Nitsan Wakart's 'JCTools', Java Concurrency Tools for the JVM, + which can be obtained at: + + * LICENSE: + * license/LICENSE.jctools.txt (ASL2 License) + * HOMEPAGE: + * https://github.com/JCTools/JCTools + + This product optionally depends on 'JZlib', a re-implementation of zlib in + pure Java, which can be obtained at: + + * LICENSE: + * license/LICENSE.jzlib.txt (BSD style License) + * HOMEPAGE: + * http://www.jcraft.com/jzlib/ + + This product optionally depends on 'Compress-LZF', a Java library for encoding and + decoding data in LZF format, written by Tatu Saloranta. It can be obtained at: + + * LICENSE: + * license/LICENSE.compress-lzf.txt (Apache License 2.0) + * HOMEPAGE: + * https://github.com/ning/compress + + This product optionally depends on 'lz4', a LZ4 Java compression + and decompression library written by Adrien Grand. It can be obtained at: + + * LICENSE: + * license/LICENSE.lz4.txt (Apache License 2.0) + * HOMEPAGE: + * https://github.com/jpountz/lz4-java + + This product optionally depends on 'lzma-java', a LZMA Java compression + and decompression library, which can be obtained at: + + * LICENSE: + * license/LICENSE.lzma-java.txt (Apache License 2.0) + * HOMEPAGE: + * https://github.com/jponge/lzma-java + + This product contains a modified portion of 'jfastlz', a Java port of FastLZ compression + and decompression library written by William Kinney. It can be obtained at: + + * LICENSE: + * license/LICENSE.jfastlz.txt (MIT License) + * HOMEPAGE: + * https://code.google.com/p/jfastlz/ + + This product contains a modified portion of and optionally depends on 'Protocol Buffers', Google's data + interchange format, which can be obtained at: + + * LICENSE: + * license/LICENSE.protobuf.txt (New BSD License) + * HOMEPAGE: + * https://github.com/google/protobuf + + This product optionally depends on 'Bouncy Castle Crypto APIs' to generate + a temporary self-signed X.509 certificate when the JVM does not provide the + equivalent functionality. It can be obtained at: + + * LICENSE: + * license/LICENSE.bouncycastle.txt (MIT License) + * HOMEPAGE: + * http://www.bouncycastle.org/ + + This product optionally depends on 'Snappy', a compression library produced + by Google Inc, which can be obtained at: + + * LICENSE: + * license/LICENSE.snappy.txt (New BSD License) + * HOMEPAGE: + * https://github.com/google/snappy + + This product optionally depends on 'JBoss Marshalling', an alternative Java + serialization API, which can be obtained at: + + * LICENSE: + * license/LICENSE.jboss-marshalling.txt (GNU LGPL 2.1) + * HOMEPAGE: + * http://www.jboss.org/jbossmarshalling + + This product optionally depends on 'Caliper', Google's micro- + benchmarking framework, which can be obtained at: + + * LICENSE: + * license/LICENSE.caliper.txt (Apache License 2.0) + * HOMEPAGE: + * https://github.com/google/caliper + + This product optionally depends on 'Apache Log4J', a logging framework, which + can be obtained at: + + * LICENSE: + * license/LICENSE.log4j.txt (Apache License 2.0) + * HOMEPAGE: + * http://logging.apache.org/log4j/ + + This product optionally depends on 'Aalto XML', an ultra-high performance + non-blocking XML processor, which can be obtained at: + + * LICENSE: + * license/LICENSE.aalto-xml.txt (Apache License 2.0) + * HOMEPAGE: + * http://wiki.fasterxml.com/AaltoHome + + This product contains a modified version of 'HPACK', a Java implementation of + the HTTP/2 HPACK algorithm written by Twitter. It can be obtained at: + + * LICENSE: + * license/LICENSE.hpack.txt (Apache License 2.0) + * HOMEPAGE: + * https://github.com/twitter/hpack + + This product contains a modified portion of 'Apache Commons Lang', a Java library + provides utilities for the java.lang API, which can be obtained at: + + * LICENSE: + * license/LICENSE.commons-lang.txt (Apache License 2.0) + * HOMEPAGE: + * https://commons.apache.org/proper/commons-lang/ + + This product contains a forked and modified version of Tomcat Native + + * LICENSE: + * ASL2 + * HOMEPAGE: + * http://tomcat.apache.org/native-doc/ + * https://svn.apache.org/repos/asf/tomcat/native/ + + (ASLv2) Objenesis + The following NOTICE information applies: + Objenesis + Copyright 2006-2013 Joe Walnes, Henri Tremblay, Leonardo Mesquita + +************************ +Eclipse Public License 1.0 +************************ + +The following binary components are provided under the Eclipse Public License 1.0. See project link for details. + + (EPL 2.0)(GPL 2)(LGPL 2.1) JNR Posix ( jnr.posix ) https://github.com/jnr/jnr-posix/blob/master/LICENSE.txt + (EPL 1.0)(LGPL 2.1) Logback Classic (ch.qos.logback:logback-classic:jar:1.2.6 - http://logback.qos.ch/) + (EPL 1.0)(LGPL 2.1) Logback Core (ch.qos.logback:logback-core:jar:1.2.6 - http://logback.qos.ch/) + +************************ +Test-Only Resources +************************ + +The following are not distributed as part of this module's binary (.nar) build, but are present in the +source tree under nifi-cassandra-session-provider-service/src/test/resources/ and are copied, with +explanatory comments trimmed, from the stock cassandra.yaml shipped in the corresponding Apache Cassandra +release: + + - cassandra-base-config/cassandra.yaml (from Apache Cassandra 5.0) + - cassandra-auth-config-4.1/cassandra.yaml (from Apache Cassandra 4.1) + - cassandra-auth-config-3.11/cassandra.yaml (from Apache Cassandra 3.11) + + (ASLv2) Apache Cassandra + The following NOTICE information applies: + Apache Cassandra + Copyright 2009- The Apache Software Foundation + + This product includes software developed by The Apache Software + Foundation (http://www.apache.org/). diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/pom.xml b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/pom.xml new file mode 100644 index 000000000000..c39f6c44e919 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/pom.xml @@ -0,0 +1,125 @@ + + + + + nifi-cql-bundle + org.apache.nifi + 2.12.0-SNAPSHOT + + 4.0.0 + + nifi-cassandra-session-provider-service + jar + + + 4.19.3 + + + + + org.apache.nifi + nifi-api + + + org.apache.nifi + nifi-utils + + + org.apache.nifi + nifi-cql-services-api + 2.12.0-SNAPSHOT + provided + + + + org.apache.cassandra + java-driver-core + ${driver.version} + + + + + org.apache.cassandra + java-driver-query-builder + ${driver.version} + + + + org.apache.nifi + nifi-ssl-context-service-api + + + org.apache.nifi + nifi-framework-api + + + org.apache.nifi + nifi-mock + test + + + + org.apache.avro + avro + + + + org.testcontainers + testcontainers-cassandra + test + + + org.testcontainers + testcontainers-junit-jupiter + ${testcontainers.version} + test + + + org.apache.nifi + nifi-security-cert-builder + 2.12.0-SNAPSHOT + test + + + org.apache.nifi + nifi-ssl-context-service + 2.12.0-SNAPSHOT + test + + + org.apache.nifi + nifi-cql-it-common + 2.12.0-SNAPSHOT + tests + test + + + org.apache.nifi + nifi-record + compile + + + org.apache.nifi + nifi-avro-record-utils + 2.12.0-SNAPSHOT + compile + + + diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/CassandraCQLExecutionService.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/CassandraCQLExecutionService.java new file mode 100644 index 000000000000..b626c9d341b4 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/CassandraCQLExecutionService.java @@ -0,0 +1,1272 @@ +/* + * 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.service.cassandra; + +import com.datastax.oss.driver.api.core.AllNodesFailedException; +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.CqlSession; +import com.datastax.oss.driver.api.core.CqlSessionBuilder; +import com.datastax.oss.driver.api.core.DriverTimeoutException; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.config.DriverConfigLoader; +import com.datastax.oss.driver.api.core.cql.BatchStatement; +import com.datastax.oss.driver.api.core.cql.BatchStatementBuilder; +import com.datastax.oss.driver.api.core.cql.BatchType; +import com.datastax.oss.driver.api.core.cql.BoundStatement; +import com.datastax.oss.driver.api.core.cql.ColumnDefinitions; +import com.datastax.oss.driver.api.core.cql.PreparedStatement; +import com.datastax.oss.driver.api.core.cql.ResultSet; +import com.datastax.oss.driver.api.core.cql.Row; +import com.datastax.oss.driver.api.core.cql.SimpleStatement; +import com.datastax.oss.driver.api.core.cql.SimpleStatementBuilder; +import com.datastax.oss.driver.api.core.data.UdtValue; +import com.datastax.oss.driver.api.core.metadata.schema.ColumnMetadata; +import com.datastax.oss.driver.api.core.metadata.schema.KeyspaceMetadata; +import com.datastax.oss.driver.api.core.metadata.schema.TableMetadata; +import com.datastax.oss.driver.api.core.servererrors.QueryExecutionException; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.DataTypes; +import com.datastax.oss.driver.api.core.type.ListType; +import com.datastax.oss.driver.api.core.type.MapType; +import com.datastax.oss.driver.api.core.type.SetType; +import com.datastax.oss.driver.api.core.type.TupleType; +import com.datastax.oss.driver.api.core.type.UserDefinedType; +import com.datastax.oss.driver.api.core.type.codec.registry.MutableCodecRegistry; +import com.datastax.oss.driver.api.querybuilder.QueryBuilder; +import com.datastax.oss.driver.api.querybuilder.delete.DeleteSelection; +import com.datastax.oss.driver.api.querybuilder.insert.Insert; +import com.datastax.oss.driver.api.querybuilder.insert.InsertInto; +import com.datastax.oss.driver.api.querybuilder.insert.RegularInsert; +import com.datastax.oss.driver.api.querybuilder.relation.Relation; +import com.datastax.oss.driver.api.querybuilder.update.Update; +import com.datastax.oss.driver.api.querybuilder.update.UpdateStart; +import com.datastax.oss.driver.api.querybuilder.update.UpdateWithAssignments; +import org.apache.avro.Schema; +import org.apache.avro.SchemaBuilder; +import org.apache.commons.lang3.StringUtils; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnDisabled; +import org.apache.nifi.annotation.lifecycle.OnEnabled; +import org.apache.nifi.avro.AvroTypeUtil; +import org.apache.nifi.components.ConfigVerificationResult; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.controller.VerifiableControllerService; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.record.path.FieldValue; +import org.apache.nifi.record.path.RecordPath; +import org.apache.nifi.record.path.RecordPathResult; +import org.apache.nifi.serialization.record.MapRecord; +import org.apache.nifi.serialization.record.Record; +import org.apache.nifi.serialization.record.RecordSchema; +import org.apache.nifi.serialization.record.util.DataTypeUtils; +import org.apache.nifi.service.cassandra.mapping.CassandraUdtSchemaMapper; +import org.apache.nifi.service.cassandra.mapping.CharacterCodec; +import org.apache.nifi.service.cassandra.mapping.FlexibleBigIntCodec; +import org.apache.nifi.service.cassandra.mapping.FlexibleBooleanCodec; +import org.apache.nifi.service.cassandra.mapping.FlexibleCounterCodec; +import org.apache.nifi.service.cassandra.mapping.FlexibleDateCodec; +import org.apache.nifi.service.cassandra.mapping.FlexibleDoubleCodec; +import org.apache.nifi.service.cassandra.mapping.FlexibleFloatCodec; +import org.apache.nifi.service.cassandra.mapping.FlexibleIntCodec; +import org.apache.nifi.service.cassandra.mapping.FlexibleSmallIntCodec; +import org.apache.nifi.service.cassandra.mapping.FlexibleTimeCodec; +import org.apache.nifi.service.cassandra.mapping.FlexibleTinyIntCodec; +import org.apache.nifi.service.cassandra.mapping.JavaSQLDateCodec; +import org.apache.nifi.service.cassandra.mapping.JavaSQLTimeCodec; +import org.apache.nifi.service.cassandra.mapping.JavaSQLTimestampCodec; +import org.apache.nifi.service.cql.api.constants.CqlBatchType; +import org.apache.nifi.service.cql.api.constants.PrimaryKeyFieldType; +import org.apache.nifi.service.cql.api.constants.UpdateMethod; +import org.apache.nifi.service.cql.api.exception.QueryFailureException; +import org.apache.nifi.service.cql.api.lookup.CqlCell; +import org.apache.nifi.service.cql.api.lookup.CqlRow; +import org.apache.nifi.service.cql.api.lookup.CqlStatementResult; +import org.apache.nifi.service.cql.api.lookup.impl.StandardCqlCell; +import org.apache.nifi.service.cql.api.lookup.impl.StandardCqlRow; +import org.apache.nifi.service.cql.api.lookup.impl.StandardCqlStatementResult; +import org.apache.nifi.service.cql.api.metadata.PrimaryKey; +import org.apache.nifi.service.cql.api.metadata.PrimaryKeyIdentifier; +import org.apache.nifi.service.cql.api.metadata.PrimaryKeyMetadata; +import org.apache.nifi.service.cql.api.metadata.QualifiedTableName; +import org.apache.nifi.service.cql.api.service.AbstractCQLExecutionService; +import org.apache.nifi.service.cql.api.service.CQLExecutionService; +import org.apache.nifi.service.cql.api.service.CQLQueryCallback; +import org.apache.nifi.service.cql.api.service.ContactPoints; +import org.apache.nifi.service.cql.api.service.QueryOverrides; +import org.apache.nifi.service.cql.api.service.WriteOverrides; +import org.apache.nifi.ssl.SSLContextProvider; + +import java.io.File; +import java.net.InetSocketAddress; +import java.time.Duration; +import java.time.Instant; +import java.time.format.DateTimeParseException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashMap; +import java.util.Iterator; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import javax.net.ssl.SSLContext; + +@Tags({"cassandra", "cql", "database", "connection", "session", "pooling"}) +@CapabilityDescription("Provides a CQL connection session for CQL processors and controller services to work with Apache Cassandra. " + + "Use this service for Apache Cassandra clusters; use ScyllaDBCQLExecutionService for ScyllaDB.") +@SeeAlso(classNames = { + "org.apache.nifi.processors.cql.PutCQLRecord", + "org.apache.nifi.processors.cql.ExecuteCQLQueryRecord", + "org.apache.nifi.service.cql.cache.CQLDistributedMapCache", + "org.apache.nifi.service.scylladb.ScyllaDBCQLExecutionService" +}) +public class CassandraCQLExecutionService extends AbstractCQLExecutionService implements CQLExecutionService, VerifiableControllerService { + + /** + * The column a conditional statement's outcome arrives in. Named with brackets by the protocol precisely + * because no CQL identifier may contain them, so it can never collide with a real column. + */ + private static final String APPLIED_COLUMN = "[applied]"; + + /** + * Bind marker name for an explicit per-record write timestamp (see {@link WriteOverrides#timestampField()}). + * Namespaced to make an accidental collision with a real column name effectively impossible. + */ + private static final String WRITE_TIMESTAMP_BIND_MARKER = "nifi_write_timestamp"; + + private static final boolean VALUE_LOGGING_ENABLED = Boolean.getBoolean("ENABLE_CQL_VALUE_LOGGING"); + + private CqlSession cassandraSession; + + private int pageSize; + + private Duration defaultTtl; + + public static final List PROPERTY_DESCRIPTORS = List.of( + CONTACT_POINTS, + DATACENTER, + KEYSPACE, + USERNAME, + PASSWORD, + PROP_SSL_CONTEXT_SERVICE, + FETCH_SIZE, + READ_TIMEOUT, + CONNECT_TIMEOUT, + CONSISTENCY_LEVEL, + COMPRESSION_TYPE, + DEFAULT_TTL, + DRIVER_CONFIGURATION_FILE + ); + + @Override + public List getSupportedPropertyDescriptors() { + return PROPERTY_DESCRIPTORS; + } + + @OnEnabled + public void onEnabled(final ConfigurationContext context) { + tableMetadataCache = new ConcurrentHashMap<>(); + connectToCassandra(context); + } + + @OnDisabled + public void onDisabled() { + if (cassandraSession != null) { + cassandraSession.close(); + cassandraSession = null; + } + } + + private Map tableMetadataCache; + private String defaultKeyspace; + + /** + * Returns the given table name with the connection's default keyspace filled in when it is unqualified, so + * metadata lookups have the explicit keyspace they require. + */ + private QualifiedTableName resolveKeyspace(QualifiedTableName table) { + return table.isQualified() ? table : new QualifiedTableName(defaultKeyspace, table.table()); + } + + private void connectToCassandra(ConfigurationContext context) { + if (cassandraSession == null) { + this.pageSize = context.getProperty(FETCH_SIZE).evaluateAttributeExpressions().asInteger(); + + final PropertyValue defaultTtlProperty = context.getProperty(DEFAULT_TTL).evaluateAttributeExpressions(); + this.defaultTtl = defaultTtlProperty.isSet() ? defaultTtlProperty.asDuration() : null; + + final CqlSession cqlSession = buildSession(context); + + try { + MutableCodecRegistry codecRegistry = + (MutableCodecRegistry) cqlSession.getContext().getCodecRegistry(); + + codecRegistry.register(new JavaSQLTimestampCodec()); + codecRegistry.register(new JavaSQLDateCodec()); + codecRegistry.register(new JavaSQLTimeCodec()); + codecRegistry.register(new CharacterCodec()); + codecRegistry.register(new FlexibleCounterCodec()); + codecRegistry.register(new FlexibleBooleanCodec()); + codecRegistry.register(new FlexibleTinyIntCodec()); + codecRegistry.register(new FlexibleSmallIntCodec()); + codecRegistry.register(new FlexibleIntCodec()); + codecRegistry.register(new FlexibleBigIntCodec()); + codecRegistry.register(new FlexibleFloatCodec()); + codecRegistry.register(new FlexibleDoubleCodec()); + codecRegistry.register(new FlexibleDateCodec()); + codecRegistry.register(new FlexibleTimeCodec()); + } catch (Exception ex) { + cqlSession.close(); + throw new ProcessException("There was an error registering conversion codecs.", ex); + } + + cassandraSession = cqlSession; + } + } + + private CqlSession buildSession(final ConfigurationContext context) { + return buildSession(context, true); + } + + /** + * @param bindKeyspace when {@code false}, the session is opened without binding the configured keyspace. + * {@link #verify(ConfigurationContext, ComponentLog, Map)} passes {@code false} so a keyspace that does + * not exist is reported by the dedicated "Verify Keyspace" step rather than failing session creation + * outright - the driver rejects an unknown keyspace during {@code CqlSession.build()}. + */ + private CqlSession buildSession(final ConfigurationContext context, final boolean bindKeyspace) { + final String consistencyLevel = context.getProperty(CONSISTENCY_LEVEL).getValue(); + final String compression = context.getProperty(COMPRESSION_TYPE).getValue(); + final String contactPointList = context.getProperty(CONTACT_POINTS).evaluateAttributeExpressions().getValue(); + + List contactPoints = getContactPoints(contactPointList); + + // Set up the client for secure (SSL/TLS communications) if configured to do so + final SSLContextProvider sslContextProvider = + context.getProperty(PROP_SSL_CONTEXT_SERVICE).asControllerService(SSLContextProvider.class); + final SSLContext sslContext; + + if (sslContextProvider == null) { + sslContext = null; + } else { + sslContext = sslContextProvider.createContext(); + } + + final String username; + final String password; + PropertyValue usernameProperty = context.getProperty(USERNAME).evaluateAttributeExpressions(); + PropertyValue passwordProperty = context.getProperty(PASSWORD).evaluateAttributeExpressions(); + + if (usernameProperty != null && passwordProperty != null) { + username = usernameProperty.getValue(); + password = passwordProperty.getValue(); + } else { + username = null; + password = null; + } + + final Duration readTimeout = context.getProperty(READ_TIMEOUT).evaluateAttributeExpressions().asDuration(); + final Duration connectTimeout = context.getProperty(CONNECT_TIMEOUT).evaluateAttributeExpressions().asDuration(); + + final String datacenter = context.getProperty(DATACENTER).evaluateAttributeExpressions().getValue(); + + final String sessionKeyspace = context.getProperty(KEYSPACE).evaluateAttributeExpressions().getValue(); + this.defaultKeyspace = sessionKeyspace; + + final DriverConfigLoader propertyBasedLoader = + DriverConfigLoader.programmaticBuilder() + .withDuration(DefaultDriverOption.CONNECTION_CONNECT_TIMEOUT, connectTimeout) + .withDuration(DefaultDriverOption.REQUEST_TIMEOUT, readTimeout) + .withString(DefaultDriverOption.REQUEST_CONSISTENCY, consistencyLevel) + .withString(DefaultDriverOption.PROTOCOL_COMPRESSION, compression) + .build(); + + final DriverConfigLoader loader = buildConfigLoader(context, propertyBasedLoader); + + CqlSessionBuilder builder = CqlSession.builder() + .addContactPoints(contactPoints); + + if (StringUtils.isNotBlank(username) && StringUtils.isNotBlank(password)) { + builder = builder.withAuthCredentials(username, password); + } + + builder = builder + .withSslContext(sslContext) + .withLocalDatacenter(datacenter) + .withConfigLoader(loader); + + if (bindKeyspace) { + builder = builder.withKeyspace(sessionKeyspace); + } + + return builder.build(); + } + + /** + * Builds the {@link DriverConfigLoader} used to open the session. When a Driver Configuration File is + * configured, it is composed as the primary source ahead of {@code propertyBasedLoader}, so any option + * defined in the file overrides the corresponding connection property, while options absent from the + * file still fall back to the property-derived configuration. The file uses the driver's standard + * typesafe-config format, which the ScyllaDB Java Driver reads identically, so this method requires no + * override for ScyllaDB compatibility. + */ + protected DriverConfigLoader buildConfigLoader(final ConfigurationContext context, final DriverConfigLoader propertyBasedLoader) { + final PropertyValue configFileProperty = context.getProperty(DRIVER_CONFIGURATION_FILE); + + if (configFileProperty == null || !configFileProperty.isSet()) { + return propertyBasedLoader; + } + + final String configFilePath = configFileProperty.evaluateAttributeExpressions().getValue(); + final DriverConfigLoader fileBasedLoader = DriverConfigLoader.fromFile(new File(configFilePath)); + + return DriverConfigLoader.compose(fileBasedLoader, propertyBasedLoader); + } + + @Override + public List verify(final ConfigurationContext context, final ComponentLog verificationLogger, final Map variables) { + final List results = new ArrayList<>(); + + final String datacenter = context.getProperty(DATACENTER).evaluateAttributeExpressions().getValue(); + + CqlSession session = null; + try { + // Opened without binding the keyspace: the driver rejects an unknown keyspace during build(), + // which would collapse a missing-keyspace failure into this step instead of the "Verify Keyspace" + // one below. Keyspace existence is checked explicitly against the session metadata further down. + session = buildSession(context, false); + results.add(new ConfigVerificationResult.Builder() + .verificationStepName("Establish Connection") + .outcome(ConfigVerificationResult.Outcome.SUCCESSFUL) + .explanation("Successfully connected to Cassandra using datacenter [" + datacenter + "]") + .build()); + } catch (final Exception e) { + verificationLogger.warn("Failed to establish Cassandra connection using datacenter [{}]: {}", datacenter, e.getMessage()); + results.add(new ConfigVerificationResult.Builder() + .verificationStepName("Establish Connection") + .outcome(ConfigVerificationResult.Outcome.FAILED) + .explanation("Failed to connect using datacenter [" + datacenter + "]: " + e.getMessage()) + .build()); + } + + if (session != null) { + try { + // CqlSessionBuilder.build() succeeds even when no node belongs to the configured local + // datacenter: the driver only filters nodes by datacenter when computing a query plan for + // an actual request. Executing a trivial statement here forces that node selection so a + // bad datacenter name is caught during verification instead of on first real query. + try { + session.execute("SELECT release_version FROM system.local"); + results.add(new ConfigVerificationResult.Builder() + .verificationStepName("Verify Datacenter") + .outcome(ConfigVerificationResult.Outcome.SUCCESSFUL) + .explanation("Found an available node in datacenter [" + datacenter + "]") + .build()); + } catch (final Exception e) { + verificationLogger.warn("No available node found in datacenter [{}]: {}", datacenter, e.getMessage()); + results.add(new ConfigVerificationResult.Builder() + .verificationStepName("Verify Datacenter") + .outcome(ConfigVerificationResult.Outcome.FAILED) + .explanation("No available node found in datacenter [" + datacenter + "]: " + e.getMessage()) + .build()); + } + + final String keyspaceName = context.getProperty(KEYSPACE).isSet() + ? context.getProperty(KEYSPACE).evaluateAttributeExpressions().getValue() : null; + + if (StringUtils.isNotBlank(keyspaceName)) { + final boolean keyspaceFound = session.getMetadata().getKeyspace(keyspaceName).isPresent(); + results.add(new ConfigVerificationResult.Builder() + .verificationStepName("Verify Keyspace") + .outcome(keyspaceFound ? ConfigVerificationResult.Outcome.SUCCESSFUL : ConfigVerificationResult.Outcome.FAILED) + .explanation(keyspaceFound + ? "Found keyspace [" + keyspaceName + "]" + : "Keyspace [" + keyspaceName + "] was not found in the cluster") + .build()); + } + } finally { + session.close(); + } + } + + return results; + } + + private List getContactPoints(String contactPointList) { + return ContactPoints.parse(contactPointList).stream() + .map(contactPoint -> new InetSocketAddress(contactPoint.host(), contactPoint.port())) + .collect(Collectors.toList()); + } + + private void cacheMetadata(QualifiedTableName table) { + final QualifiedTableName resolved = resolveKeyspace(table); + getLogger().debug("Fetching metadata for {}.{}", resolved.keyspace(), resolved.table()); + + if (!tableMetadataCache.containsKey(resolved)) { + getLogger().debug("Metadata for {}.{} not found, fetching from cluster.", resolved.keyspace(), resolved.table()); + Optional keyspaceOpt = cassandraSession.getMetadata() + .getKeyspace(resolved.keyspace()); + if (keyspaceOpt.isEmpty()) { + throw new RuntimeException("Empty keyspace metadata"); + } + + KeyspaceMetadata keyspaceMetadata = keyspaceOpt.get(); + Optional tableOpt = keyspaceMetadata.getTable(resolved.table()); + if (tableOpt.isEmpty()) { + throw new RuntimeException("Empty table metadata"); + } + + TableMetadata metadata = tableOpt.get(); + PrimaryKey tableInfo = convertTableMetadata(metadata); + + tableMetadataCache.put(resolved, tableInfo); + getLogger().debug("Metadata fetched and cached for {}.{}", resolved.keyspace(), resolved.table()); + } + } + + private PrimaryKey convertTableMetadata(TableMetadata metadata) { + List partitions = new ArrayList<>(); + List clustering = new ArrayList<>(); + + int loc = 0; + for (ColumnMetadata m : metadata.getPartitionKey()) { + partitions.add(new PrimaryKeyMetadata(m.getName().asInternal(), loc++, PrimaryKeyFieldType.PARTITION)); + } + + loc = 0; + for (ColumnMetadata m : metadata.getClusteringColumns().keySet()) { + clustering.add(new PrimaryKeyMetadata(m.getName().asInternal(), loc++, PrimaryKeyFieldType.CLUSTERING)); + } + + return new PrimaryKey(partitions, clustering); + } + + private static String valuesFragment(final Object[] values) { + return VALUE_LOGGING_ENABLED ? " with values " + Arrays.toString(values) : ""; + } + + private static String valuesFragment(final List values) { + return VALUE_LOGGING_ENABLED && values != null && !values.isEmpty() ? " with values " + values : ""; + } + + @Override + public void query(String cql, List parameters, CQLQueryCallback callback, QueryOverrides overrides) throws QueryFailureException { + if (getLogger().isDebugEnabled()) { + getLogger().debug("Executing CQL query: {}{}", cql, valuesFragment(parameters)); + } + + BoundStatement boundStatement = bindStatement(cql, parameters, overrides); + + AtomicReference schemaReference = new AtomicReference<>(); + + // The whole lifecycle of the query - the initial execute() as well as every page fetch triggered by + // hasNext()/next() as the ResultSet is iterated - can surface a QueryExecutionException, so all of it + // is covered by one try, rather than only the page-fetch calls. + try { + ResultSet results = cassandraSession.execute(boundStatement); + Iterator resultsIterator = results.iterator(); + long rowNumber = 0; + + List columnNames = new ArrayList<>(); + + while (resultsIterator.hasNext()) { + Row row = resultsIterator.next(); + + if (schemaReference.get() == null) { + Schema generatedAvroSchema = createSchema(results); + RecordSchema converted = AvroTypeUtil.createSchema(generatedAvroSchema); + schemaReference.set(converted); + } + + if (columnNames.isEmpty()) { + row.getColumnDefinitions().forEach(def -> { + columnNames.add(def.getName().toString()); + }); + } + + Map resultMap = new HashMap<>(); + + for (int x = 0; x < columnNames.size(); x++) { + resultMap.put(columnNames.get(x), row.getObject(x)); + } + + MapRecord record = new MapRecord(schemaReference.get(), resultMap); + + callback.receive(++rowNumber, record, resultsIterator.hasNext()); + } + // Each of these is worth another attempt: a coordinator/replica-level execution failure, no reachable + // node, or a client-side request timeout. Deliberately not their DriverException base type - the + // validation errors under it (SyntaxError, InvalidQueryException, UnauthorizedException) can never + // succeed on a retry, so they belong on the failure path below. + } catch (QueryExecutionException | AllNodesFailedException | DriverTimeoutException qee) { + getLogger().error("Error executing query", qee); + callback.clear(); + throw new QueryFailureException(); + } catch (Exception ex) { + callback.clear(); + throw new ProcessException("Error querying CQL", ex); + } + } + + @Override + public CqlStatementResult execute(final String cql, final List parameters, final QueryOverrides overrides) + throws QueryFailureException { + if (getLogger().isDebugEnabled()) { + getLogger().debug("Executing CQL statement: {}{}", cql, valuesFragment(parameters)); + } + + final BoundStatement boundStatement = bindStatement(cql, parameters, overrides); + + try { + final ResultSet results = cassandraSession.execute(boundStatement); + final boolean wasApplied = results.wasApplied(); + + // The outcome column is present exactly when the statement was conditional - wasApplied() alone + // cannot tell us, since it also returns true for every unconditional statement. + final boolean conditional = results.getColumnDefinitions().contains(APPLIED_COLUMN); + + // A conditional write that applied has no prior value to report, so it reports no rows. The + // backends disagree about what they send in that case: Cassandra returns the outcome column + // alone, ScyllaDB returns the other columns alongside it. Normalizing here is the whole point of + // having one API over both - otherwise callers would find rows().isEmpty() backend-dependent. + if (conditional && wasApplied) { + return new StandardCqlStatementResult(true, List.of()); + } + + final List rows = new ArrayList<>(); + for (final Row row : results) { + rows.add(new StandardCqlRow(toCells(row))); + } + + return new StandardCqlStatementResult(wasApplied, rows); + } catch (QueryExecutionException | AllNodesFailedException | DriverTimeoutException qee) { + getLogger().error("Error executing statement", qee); + throw new QueryFailureException(); + } catch (Exception ex) { + throw new ProcessException("Error executing CQL", ex); + } + } + + /** + * Converts one driver row into cells, dropping the {@code [applied]} conditional-outcome column and marking + * any column whose CQL type has no JDK representation as unsupported. + */ + private List toCells(final Row row) { + final ColumnDefinitions definitions = row.getColumnDefinitions(); + final List cells = new ArrayList<>(definitions.size()); + + for (int index = 0; index < definitions.size(); index++) { + final String columnName = definitions.get(index).getName().toString(); + if (APPLIED_COLUMN.equals(columnName)) { + continue; + } + + final DataType columnType = definitions.get(index).getType(); + cells.add(hasJdkRepresentation(columnType) + ? new StandardCqlCell(columnName, row.getObject(index)) + : StandardCqlCell.unsupported(columnName, columnType.asCql(true, false))); + } + + return cells; + } + + /** + * Reports whether a CQL type decodes to a JDK type rather than a driver class, recursing through collection + * types so a {@code list} is rejected for its element type. + */ + private static boolean hasJdkRepresentation(final DataType type) { + return switch (type) { + case ListType list -> hasJdkRepresentation(list.getElementType()); + case SetType set -> hasJdkRepresentation(set.getElementType()); + case MapType map -> hasJdkRepresentation(map.getKeyType()) && hasJdkRepresentation(map.getValueType()); + default -> !(type instanceof UserDefinedType || type instanceof TupleType || DataTypes.DURATION.equals(type)); + }; + } + + /** + * Builds and binds a statement, applying any per-call fetch size and timeout overrides. Shared by + * {@link #query} and {@link #execute}, which differ only in what they do with the result. + */ + private BoundStatement bindStatement(final String cql, final List parameters, final QueryOverrides overrides) { + SimpleStatementBuilder statementBuilder = SimpleStatement.builder(cql) + .setPageSize(resolveFetchSize(overrides, pageSize)); + + final Duration timeoutOverride = resolveTimeoutOverride(overrides); + if (timeoutOverride != null) { + statementBuilder = statementBuilder.setTimeout(timeoutOverride); + } + + final SimpleStatement statement = statementBuilder.build(); + final PreparedStatement preparedStatement = cassandraSession.prepare(statement); + + return parameters != null && !parameters.isEmpty() + ? preparedStatement.bind(toBindValues(preparedStatement, parameters)) + : preparedStatement.bind(); + } + + static int resolveFetchSize(final QueryOverrides overrides, final int defaultFetchSize) { + return overrides != null && overrides.fetchSize() != null ? overrides.fetchSize() : defaultFetchSize; + } + + static Duration resolveTimeoutOverride(final QueryOverrides overrides) { + return overrides != null ? overrides.timeout() : null; + } + + static BatchType toDriverBatchType(final CqlBatchType batchType) { + return switch (batchType) { + case LOGGED -> BatchType.LOGGED; + case UNLOGGED -> BatchType.UNLOGGED; + case COUNTER -> BatchType.COUNTER; + }; + } + + static Integer resolveTtlSeconds(final WriteOverrides overrides, final Duration defaultTtl) { + final Duration ttl = overrides != null && overrides.ttl() != null ? overrides.ttl() : defaultTtl; + return ttl != null ? Math.toIntExact(ttl.toSeconds()) : null; + } + + static boolean hasTimestampOverride(final WriteOverrides overrides) { + return overrides != null && overrides.timestampField() != null; + } + + /** + * Converts a record field's value into epoch microseconds, the unit Cassandra/ScyllaDB expect for an + * explicit write timestamp (distinct from the millisecond precision of the CQL {@code timestamp} type). + */ + static long toEpochMicros(final Object value) { + final Instant instant; + if (value instanceof Instant instantValue) { + instant = instantValue; + } else if (value instanceof java.util.Date dateValue) { + instant = dateValue.toInstant(); + } else if (value instanceof Number numberValue) { + instant = Instant.ofEpochMilli(numberValue.longValue()); + } else { + throw new IllegalArgumentException("Cannot derive a write timestamp from a value of type " + + (value == null ? "null" : value.getClass().getName())); + } + + return instant.getEpochSecond() * 1_000_000L + instant.getNano() / 1_000L; + } + + protected GeneratedResult generateInsert(QualifiedTableName table, RecordSchema schema, + Map primaryKeyOverrides, + Integer ttlSeconds, boolean includeTimestampMarker) { + InsertInto insertQuery = table.isQualified() + ? QueryBuilder.insertInto(table.keyspace(), table.table()) + : QueryBuilder.insertInto(table.table()); + List keys = new ArrayList<>(); + + RegularInsert regularInsert = null; + for (String fieldName : columnNamesIncludingOverrideOnly(table, schema, primaryKeyOverrides)) { + if (regularInsert == null) { + regularInsert = insertQuery.value(fieldName, QueryBuilder.bindMarker(fieldName)); + } else { + regularInsert = regularInsert.value(fieldName, QueryBuilder.bindMarker(fieldName)); + } + + keys.add(fieldName); + } + + if (regularInsert == null) { + throw new ProcessException("Could not build an insert statement from the supplied record"); + } + + Insert insert = regularInsert; + if (ttlSeconds != null) { + insert = insert.usingTtl(ttlSeconds); + } + if (includeTimestampMarker) { + // A separate named bind marker (set by name on the bound statement, not part of "keys") rather than + // a literal, since - unlike TTL - each record in a batch can carry a different write timestamp. + insert = insert.usingTimestamp(QueryBuilder.bindMarker(WRITE_TIMESTAMP_BIND_MARKER)); + } + + return new GeneratedResult(insert.build(), keys); + } + + /** + * Returns every schema field name in schema order, followed by any primary key override target column for + * this table that has no same-named schema field. + */ + private List columnNamesIncludingOverrideOnly(QualifiedTableName table, RecordSchema schema, + Map primaryKeyOverrides) { + final List columnNames = new ArrayList<>(schema.getFieldNames()); + for (String overrideFieldName : overrideFieldNamesForTable(table, primaryKeyOverrides)) { + if (!columnNames.contains(overrideFieldName)) { + columnNames.add(overrideFieldName); + } + } + return columnNames; + } + + /** + * The field names of every primary key override configured for this specific table, resolving an + * unqualified {@code table} against {@link #defaultKeyspace} the same way {@link #getRecordPathOverride} + * does. + */ + private Set overrideFieldNamesForTable(QualifiedTableName table, Map primaryKeyOverrides) { + final QualifiedTableName resolved = resolveKeyspace(table); + return primaryKeyOverrides.keySet().stream() + .filter(identifier -> identifier.keyspace().equals(resolved.keyspace()) && identifier.tableName().equals(resolved.table())) + .map(PrimaryKeyIdentifier::fieldName) + .collect(Collectors.toCollection(LinkedHashSet::new)); + } + + @Override + public void insert(QualifiedTableName table, Record record, Map primaryKeyOverrides, + WriteOverrides overrides) throws QueryFailureException { + final boolean useTimestampOverride = hasTimestampOverride(overrides); + GeneratedResult result = generateInsert(table, record.getSchema(), primaryKeyOverrides, resolveTtlSeconds(overrides, defaultTtl), useTimestampOverride); + PreparedStatement preparedStatement = cassandraSession.prepare(result.statement); + + Object[] values = getBindValues(table, primaryKeyOverrides, record, preparedStatement, result.keysUsed); + if (useTimestampOverride) { + setTimestampBindValue(values, preparedStatement, record, overrides); + } + + if (getLogger().isDebugEnabled()) { + getLogger().debug("Executing CQL insert: {}{}", result.statement.getQuery(), valuesFragment(values)); + } + + BoundStatement boundStatement = preparedStatement.bind(values); + + try { + cassandraSession.execute(boundStatement); + } catch (QueryExecutionException | AllNodesFailedException | DriverTimeoutException qee) { + getLogger().error("Error executing insert", qee); + throw new QueryFailureException(); + } + } + + @Override + public void insert(QualifiedTableName table, List records, Map primaryKeyOverrides, + CqlBatchType batchType, WriteOverrides overrides) throws QueryFailureException { + if (records == null || records.isEmpty()) { + return; + } + + final boolean useTimestampOverride = hasTimestampOverride(overrides); + BatchStatementBuilder builder = BatchStatement.builder(toDriverBatchType(batchType)); + GeneratedResult result = generateInsert(table, records.get(0).getSchema(), primaryKeyOverrides, resolveTtlSeconds(overrides, defaultTtl), useTimestampOverride); + PreparedStatement preparedStatement = cassandraSession.prepare(result.statement); + + if (getLogger().isDebugEnabled()) { + getLogger().debug("Executing CQL {} batch insert of {} record(s): {}", batchType, records.size(), result.statement.getQuery()); + } + + int recordNumber = 0; + for (Record record : records) { + recordNumber++; + Object[] values = getBindValues(table, primaryKeyOverrides, record, preparedStatement, result.keysUsed); + if (useTimestampOverride) { + setTimestampBindValue(values, preparedStatement, record, overrides); + } + + if (VALUE_LOGGING_ENABLED && getLogger().isDebugEnabled()) { + getLogger().debug("Batch insert record {}{}", recordNumber, valuesFragment(values)); + } + + builder.addStatement(preparedStatement.bind(values)); + } + + try { + cassandraSession.execute(builder.build()); + } catch (QueryExecutionException | AllNodesFailedException | DriverTimeoutException qee) { + getLogger().error("Error executing batch insert", qee); + throw new QueryFailureException(); + } + } + + @Override + public String getTransitUrl(QualifiedTableName tableName) { + final String qualifiedName = tableName.isQualified() + ? tableName.keyspace() + "." + tableName.table() + : tableName.table(); + return "cassandra://" + cassandraSession.getMetadata().getClusterName() + "." + qualifiedName; + } + + /** + * Creates an Avro schema from the given result set. The metadata (column definitions, data types, etc.) is used + * to determine a schema for Avro. + * + * @param rs The result set from which an Avro schema will be created + * @return An Avro schema corresponding to the given result set's metadata + */ + public static Schema createSchema(final ResultSet rs) { + final ColumnDefinitions columnDefinitions = rs.getColumnDefinitions(); + final int nrOfColumns = (columnDefinitions == null ? 0 : columnDefinitions.size()); + String tableName = "NiFi_Cassandra_Query_Record"; + if (nrOfColumns > 0) { + String tableNameFromMeta = columnDefinitions.get(0).getTable().toString(); + if (!StringUtils.isBlank(tableNameFromMeta)) { + tableName = tableNameFromMeta; + } + } + + final SchemaBuilder.FieldAssembler builder = SchemaBuilder.record(tableName).namespace("any.data").fields(); + if (columnDefinitions != null) { + final Map udtSchemaCache = new HashMap<>(); + for (int i = 0; i < nrOfColumns; i++) { + DataType dataType = columnDefinitions.get(i).getType(); + if (dataType == null) { + throw new IllegalArgumentException("No data type for column[" + i + "] with name " + + columnDefinitions.get(i).getName()); + } + + builder.name(columnDefinitions.get(i).getName().toString()) + .type(CassandraUdtSchemaMapper.toAvroSchema(dataType, udtSchemaCache)) + .noDefault(); + } + } + return builder.endRecord(); + } + + protected GeneratedResult generateDelete(QualifiedTableName cassandraTable, Record record, + Map primaryKeyOverrides, List deleteKeyNames) { + RecordSchema schema = record.getSchema(); + + if (deleteKeyNames == null || deleteKeyNames.isEmpty()) { + throw new IllegalArgumentException("No delete keys were specified"); + } + + // Verify every delete key is either a real record field or covered by a primary key override for this + // table, matching generateUpdate: a key resolved purely by RecordPath has no same-named record field, + // but getBindValues can still resolve its value from the override alone. + final Set overrideFieldNames = overrideFieldNamesForTable(cassandraTable, primaryKeyOverrides); + for (String deleteKey : deleteKeyNames) { + if (!schema.getFieldNames().contains(deleteKey) && !overrideFieldNames.contains(deleteKey)) { + throw new IllegalArgumentException("Delete key '" + deleteKey + "' is not present in the record schema"); + } + } + + DeleteSelection deleteSelection = cassandraTable.isQualified() + ? QueryBuilder.deleteFrom(cassandraTable.keyspace(), cassandraTable.table()) + : QueryBuilder.deleteFrom(cassandraTable.table()); + + List whereCriteria = new ArrayList<>(); + List keysUsedInOrder = new ArrayList<>(); + + for (String fieldName : deleteKeyNames) { + whereCriteria.add(Relation.column(fieldName).isEqualTo(QueryBuilder.bindMarker(fieldName))); + keysUsedInOrder.add(fieldName); + } + + return new GeneratedResult(deleteSelection.where(whereCriteria).build(), keysUsedInOrder); + } + + protected GeneratedResult generateUpdate(QualifiedTableName cassandraTable, Record record, Map primaryKeyOverrides, + List updateKeyNames, UpdateMethod updateMethod, Integer ttlSeconds, boolean includeTimestampMarker) { + RecordSchema schema = record.getSchema(); + + List keysUsedInOrder = new ArrayList<>(); + + if (updateKeyNames == null || updateKeyNames.isEmpty()) { + throw new IllegalArgumentException("No Update Keys were specified"); + } + + // Verify every update key is either a real record field or covered by a primary key override for + // this table - an update key with neither, such as a date column derived from a timestamp field via + // a format() RecordPath override, would otherwise be rejected here even though getBindValues can + // resolve its value perfectly well from the override alone. + final Set overrideFieldNames = overrideFieldNamesForTable(cassandraTable, primaryKeyOverrides); + for (String updateKey : updateKeyNames) { + if (!schema.getFieldNames().contains(updateKey) && !overrideFieldNames.contains(updateKey)) { + throw new IllegalArgumentException("Update key '" + updateKey + "' is not present in the record schema"); + } + } + + UpdateStart updateQueryStart = cassandraTable.isQualified() + ? QueryBuilder.update(cassandraTable.keyspace(), cassandraTable.table()) + : QueryBuilder.update(cassandraTable.table()); + + // Cassandra/ScyllaDB do not support a TTL or a custom write timestamp on counter columns, so both are + // only applied for SET updates. + if (ttlSeconds != null && updateMethod == UpdateMethod.SET) { + updateQueryStart = updateQueryStart.usingTtl(ttlSeconds); + } + if (includeTimestampMarker && updateMethod == UpdateMethod.SET) { + updateQueryStart = updateQueryStart.usingTimestamp(QueryBuilder.bindMarker(WRITE_TIMESTAMP_BIND_MARKER)); + } + + UpdateWithAssignments updateAssignments = null; + + List otherKeys = schema.getFieldNames().stream() + .filter(fieldName -> !updateKeyNames.contains(fieldName)) + .toList(); + + for (String fieldName : otherKeys) { + if (updateMethod == UpdateMethod.SET) { + updateAssignments = updateAssignments == null ? updateQueryStart.setColumn(fieldName, QueryBuilder.bindMarker(fieldName)) + : updateAssignments.setColumn(fieldName, QueryBuilder.bindMarker(fieldName)); + } else if (updateMethod == UpdateMethod.INCREMENT) { + updateAssignments = updateAssignments == null ? updateQueryStart.increment(fieldName, QueryBuilder.bindMarker(fieldName)) + : updateAssignments.increment(fieldName, QueryBuilder.bindMarker(fieldName)); + } else if (updateMethod == UpdateMethod.DECREMENT) { + updateAssignments = updateAssignments == null ? updateQueryStart.decrement(fieldName, QueryBuilder.bindMarker(fieldName)) + : updateAssignments.decrement(fieldName, QueryBuilder.bindMarker(fieldName)); + } else { + throw new IllegalArgumentException("Update Method '" + updateMethod + "' is not valid."); + } + + keysUsedInOrder.add(fieldName); + } + + if (updateAssignments == null) { + throw new ProcessException("No update assignment found"); + } + + Update update = null; + + for (String fieldName : updateKeyNames) { + update = update == null ? updateAssignments.whereColumn(fieldName).isEqualTo(QueryBuilder.bindMarker(fieldName)) + : update.whereColumn(fieldName).isEqualTo(QueryBuilder.bindMarker(fieldName)); + keysUsedInOrder.add(fieldName); + } + + return new GeneratedResult(update.build(), keysUsedInOrder); + } + + @Override + public void delete(QualifiedTableName cassandraTable, Record record, Map primaryKeyOverrides, + List updateKeys) throws QueryFailureException { + GeneratedResult result = generateDelete(cassandraTable, record, primaryKeyOverrides, updateKeys); + PreparedStatement preparedStatement = cassandraSession.prepare(result.statement()); + + Object[] values = getBindValues(cassandraTable, primaryKeyOverrides, record, preparedStatement, result.keysUsed()); + + if (getLogger().isDebugEnabled()) { + getLogger().debug("Executing CQL delete: {}{}", result.statement().getQuery(), valuesFragment(values)); + } + + BoundStatement deleteStatement = preparedStatement.bind(values); + + try { + cassandraSession.execute(deleteStatement); + } catch (QueryExecutionException | AllNodesFailedException | DriverTimeoutException qee) { + getLogger().error("Error executing delete", qee); + throw new QueryFailureException(); + } + } + + @Override + public void update(QualifiedTableName cassandraTable, Record record, Map primaryKeyOverrides, + List updateKeys, UpdateMethod updateMethod, WriteOverrides overrides) throws QueryFailureException { + final boolean useTimestampOverride = hasTimestampOverride(overrides) && updateMethod == UpdateMethod.SET; + GeneratedResult result = generateUpdate(cassandraTable, record, primaryKeyOverrides, updateKeys, updateMethod, resolveTtlSeconds(overrides, defaultTtl), useTimestampOverride); + PreparedStatement preparedStatement = cassandraSession.prepare(result.statement()); + + Object[] values = getBindValues(cassandraTable, primaryKeyOverrides, record, preparedStatement, result.keysUsed()); + if (useTimestampOverride) { + setTimestampBindValue(values, preparedStatement, record, overrides); + } + + if (getLogger().isDebugEnabled()) { + getLogger().debug("Executing CQL {} update: {}{}", updateMethod, result.statement().getQuery(), valuesFragment(values)); + } + + BoundStatement statement = preparedStatement.bind(values); + + try { + cassandraSession.execute(statement); + } catch (QueryExecutionException | AllNodesFailedException | DriverTimeoutException qee) { + getLogger().error("Error executing update", qee); + throw new QueryFailureException(); + } + } + + /** + * Resolves each named field to its actual bind-marker position and converts its record value for that + * position's CQL type, leaving any position not named in {@code keyNamesInOrder} for the caller to set. + */ + private Object[] getBindValues(QualifiedTableName qualifiedTableName, + Map mappings, + Record record, + PreparedStatement preparedStatement, List keyNamesInOrder) { + ColumnDefinitions variableDefinitions = preparedStatement.getVariableDefinitions(); + Object[] result = new Object[variableDefinitions.size()]; + + for (String fieldName : keyNamesInOrder) { + int index = variableDefinitions.firstIndexOf(fieldName); + + RecordPath path = getRecordPathOverride(qualifiedTableName, fieldName, mappings); + Object value; + if (path != null) { + value = evaluateOverride(record, path); + } else { + value = record.getValue(fieldName); + } + DataType cqlType = variableDefinitions.get(index).getType(); + result[index] = convertForCqlType(value, cqlType); + } + + return result; + } + + /** + * Converts each positional parameter to the Java type its bind marker's declared CQL type requires, failing + * fast if the parameter count doesn't match the statement's bind markers. + */ + private Object[] toBindValues(final PreparedStatement preparedStatement, final List parameters) { + final ColumnDefinitions bindMarkers = preparedStatement.getVariableDefinitions(); + + if (bindMarkers.size() != parameters.size()) { + throw new IllegalArgumentException(String.format( + "Query declares %d bind marker(s) but %d parameter(s) were supplied", + bindMarkers.size(), parameters.size())); + } + + final Object[] values = new Object[parameters.size()]; + + for (int index = 0; index < parameters.size(); index++) { + final DataType cqlType = bindMarkers.get(index).getType(); + final Object value = parameters.get(index); + + values[index] = cqlType.equals(DataTypes.TIMESTAMP) + ? toInstant(value) + : convertForCqlType(value, cqlType); + } + + return values; + } + + /** + * Converts a value bound to a {@code timestamp} column into the {@link Instant} the driver's own codec + * expects. Unlike the other scalar types, {@code timestamp} has no {@code Flexible*Codec} registered to + * coerce a compatible-but-non-exact value, so an ISO-8601 rendering - the natural form for a parameter + * supplied as text - would otherwise fail with the driver's opaque {@code CodecNotFoundException}. + */ + static Instant toInstant(final Object value) { + if (value == null) { + return null; + } else if (value instanceof Instant instantValue) { + return instantValue; + } else if (value instanceof java.util.Date dateValue) { + return dateValue.toInstant(); + } else if (value instanceof Number numberValue) { + return Instant.ofEpochMilli(numberValue.longValue()); + } + + final String text = value.toString().trim(); + + try { + return Instant.parse(text); + } catch (final DateTimeParseException e) { + try { + return Instant.ofEpochMilli(Long.parseLong(text)); + } catch (final NumberFormatException ignored) { + throw new IllegalArgumentException(String.format( + "Value '%s' cannot be bound to a timestamp column: expected an ISO-8601 instant such as " + + "2026-08-07T14:30:00Z, or a count of milliseconds since the epoch", text), e); + } + } + } + + private Object evaluateOverride(Record record, RecordPath path) { + RecordPathResult result = path.evaluate(record); + List valueList = result.getSelectedFields().toList(); + + if (valueList.isEmpty()) { + throw new IllegalArgumentException(String.format("%s evaluated to no values.", path.getPath())); + } else if (valueList.size() > 1) { + throw new IllegalArgumentException(String.format("%s evaluated to more than one value.", path.getPath())); + } + + return valueList.get(0).getValue(); + } + + private RecordPath getRecordPathOverride(QualifiedTableName tableName, String fieldName, + Map overrides) { + final String cleanFieldName = CqlIdentifier.fromCql(fieldName).asInternal(); + final String keyspace = tableName.isQualified() ? tableName.keyspace() : defaultKeyspace; + + Optional> identifier = overrides + .entrySet() + .stream() + .filter((e) -> e.getKey().keyspace().equals(keyspace) + && e.getKey().tableName().equals(tableName.table()) + && CqlIdentifier.fromCql(e.getKey().fieldName()).asInternal().equals(cleanFieldName)) + .findFirst(); + + return identifier.map(Map.Entry::getValue).orElse(null); + } + + /** + * Fills in the slot reserved for {@link #WRITE_TIMESTAMP_BIND_MARKER} directly in the same array used for + * the single {@link PreparedStatement#bind(Object...)} call, rather than binding it separately afterward via + * a named setter on the resulting {@link BoundStatement} - the latter looked correct but did not reliably + * take effect for UPDATE statements when verified against a real cluster. + */ + private void setTimestampBindValue(Object[] values, PreparedStatement preparedStatement, Record record, WriteOverrides overrides) { + int index = preparedStatement.getVariableDefinitions().firstIndexOf(WRITE_TIMESTAMP_BIND_MARKER); + values[index] = toEpochMicros(record.getValue(overrides.timestampField())); + } + + private Object convertForCqlType(final Object value, final DataType cqlType) { + if (value == null) { + return null; + } + + // A RECORD-typed value is normally a nested Record, but DataTypeUtils#toRecord also accepts a raw Map + // as an equally valid representation, so both are converted into a UdtValue the same way here. + if (cqlType instanceof UserDefinedType udtType + && (value instanceof Record || value instanceof Map)) { + final Function fieldValueLookup = value instanceof Record nestedRecord + ? nestedRecord::getValue + : ((Map) value)::get; + + UdtValue udtValue = udtType.newValue(); + final List fieldNames = udtType.getFieldNames().stream().map(name -> name.asInternal()).toList(); + + for (int i = 0; i < fieldNames.size(); i++) { + final String fieldName = fieldNames.get(i); + final DataType fieldCqlType = udtType.getFieldTypes().get(i); + final Object fieldValue = convertForCqlType(fieldValueLookup.apply(fieldName), fieldCqlType); + try { + // A null field is set directly against its declared CQL type rather than through a codec + // lookup. Resolving the codec from the value's runtime class cannot work when there is no + // value: the lookup becomes [ <-> java.lang.Object], which no codec satisfies, + // and that made any UDT carrying an absent or optional field unwritable. + udtValue = fieldValue == null + ? udtValue.setToNull(fieldName) + : udtValue.set(fieldName, fieldValue, (Class) fieldValue.getClass()); + } catch (Exception ex) { + throw new RuntimeException(ex); + } + } + + return udtValue; + } + + // An ARRAY-typed value is normally a List, but Object[] is NiFi's own canonical representation + // (DataTypeUtils#toArray always returns one), so both are accepted here for list and set columns. + if (cqlType instanceof ListType listType && (value instanceof List || value instanceof Object[])) { + final DataType elementCqlType = listType.getElementType(); + return toElementStream(value).map(element -> convertForCqlType(element, elementCqlType)).toList(); + } + + if (cqlType instanceof SetType setType && (value instanceof Set || value instanceof Object[])) { + final DataType elementCqlType = setType.getElementType(); + return toElementStream(value).map(element -> convertForCqlType(element, elementCqlType)).collect(Collectors.toSet()); + } + + if (value instanceof Map map && cqlType instanceof MapType mapType) { + final DataType keyCqlType = mapType.getKeyType(); + final DataType valueCqlType = mapType.getValueType(); + final Map converted = new HashMap<>(); + + for (final Map.Entry entry : map.entrySet()) { + converted.put(convertForCqlType(entry.getKey(), keyCqlType), convertForCqlType(entry.getValue(), valueCqlType)); + } + + return converted; + } + + if (cqlType.equals(DataTypes.UUID)) { + return DataTypeUtils.toUUID(value); + } + + // The driver's own codec already rejects a non-version-1 UUID bound to a timeuuid column, but only + // via a confusing CodecNotFoundException ("Codec not found for requested operation: [TIMEUUID <-> + // java.util.UUID]") that reads like a driver/configuration problem rather than a data problem. + // Checking here instead produces a clear, attributable error before the value ever reaches bind(). + if (cqlType.equals(DataTypes.TIMEUUID)) { + final UUID uuid = DataTypeUtils.toUUID(value); + if (uuid.version() != 1) { + throw new IllegalArgumentException(String.format( + "Value '%s' is not a valid timeuuid: version %d, but timeuuid columns require a version 1 (time-based) UUID", + uuid, uuid.version())); + } + return uuid; + } + + return value; + } + + private static Stream toElementStream(final Object value) { + return (value instanceof Object[] array) ? Arrays.stream(array) : ((Collection) value).stream(); + } + + @Override + public void update(QualifiedTableName cassandraTable, List records, Map primaryKeyOverrides, + List updateKeys, UpdateMethod updateMethod, CqlBatchType batchType, + WriteOverrides overrides) throws QueryFailureException { + if (records == null || records.isEmpty()) { + return; + } + + final boolean useTimestampOverride = hasTimestampOverride(overrides) && updateMethod == UpdateMethod.SET; + BatchStatementBuilder builder = BatchStatement.builder(toDriverBatchType(batchType)); + + GeneratedResult result = generateUpdate(cassandraTable, records.get(0), primaryKeyOverrides, updateKeys, updateMethod, resolveTtlSeconds(overrides, defaultTtl), useTimestampOverride); + PreparedStatement preparedStatement = cassandraSession.prepare(result.statement()); + + if (getLogger().isDebugEnabled()) { + getLogger().debug("Executing CQL {} batch {} update of {} record(s): {}", batchType, updateMethod, records.size(), result.statement().getQuery()); + } + + int recordNumber = 0; + for (Record record : records) { + recordNumber++; + Object[] values = getBindValues(cassandraTable, primaryKeyOverrides, record, preparedStatement, result.keysUsed()); + if (useTimestampOverride) { + setTimestampBindValue(values, preparedStatement, record, overrides); + } + + if (VALUE_LOGGING_ENABLED && getLogger().isDebugEnabled()) { + getLogger().debug("Batch update record {}{}", recordNumber, valuesFragment(values)); + } + + builder.addStatement(preparedStatement.bind(values)); + } + + try { + cassandraSession.execute(builder.build()); + } catch (QueryExecutionException | AllNodesFailedException | DriverTimeoutException qee) { + getLogger().error("Error executing batch update", qee); + throw new QueryFailureException(); + } + } + + @Override + public PrimaryKey getMetadata(QualifiedTableName table) { + cacheMetadata(table); + return tableMetadataCache.get(resolveKeyspace(table)); + } + + @Override + public String getDefaultKeyspace() { + return this.defaultKeyspace; + } + + record GeneratedResult(SimpleStatement statement, List keysUsed) { + + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/CassandraUdtSchemaMapper.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/CassandraUdtSchemaMapper.java new file mode 100644 index 000000000000..701416d80b9a --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/CassandraUdtSchemaMapper.java @@ -0,0 +1,180 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.CqlIdentifier; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.DataTypes; +import com.datastax.oss.driver.api.core.type.ListType; +import com.datastax.oss.driver.api.core.type.MapType; +import com.datastax.oss.driver.api.core.type.SetType; +import com.datastax.oss.driver.api.core.type.UserDefinedType; +import org.apache.avro.LogicalTypes; +import org.apache.avro.Schema; +import org.apache.avro.SchemaBuilder; + +import java.util.List; +import java.util.Map; + +import static com.datastax.oss.driver.api.core.type.DataTypes.ASCII; + +/** + * Recursively converts a Cassandra CQL {@link DataType} into a nullable-union Avro {@link Schema}, mapping + * User Defined Types (including nested ones and those held in a list/set/map) to nested named Avro records. + * Any CQL type with no Avro equivalent ({@code DURATION}, {@code TUPLE}, {@code VECTOR}, custom types) falls + * back to a plain string so one exotic column never fails an entire query. + */ +public final class CassandraUdtSchemaMapper { + + private CassandraUdtSchemaMapper() { + } + + /** + * Converts the given CQL type into a nullable Avro union schema. + * + * @param cqlType the CQL type to convert + * @param udtSchemaCache UDT schemas already built in this conversion, keyed by "keyspace.typeName", so a + * UDT referenced more than once reuses one {@link Schema} instance (Avro rejects + * duplicate record names). Pass a fresh, empty map per top-level call. + */ + public static Schema toAvroSchema(final DataType cqlType, final Map udtSchemaCache) { + if (cqlType instanceof UserDefinedType udtType) { + return nullableUnion(toUdtRecordSchema(udtType, udtSchemaCache)); + } + + if (cqlType instanceof ListType listType) { + return nullableUnion(SchemaBuilder.array().items(toAvroSchema(listType.getElementType(), udtSchemaCache))); + } + + if (cqlType instanceof SetType setType) { + return nullableUnion(SchemaBuilder.array().items(toAvroSchema(setType.getElementType(), udtSchemaCache))); + } + + if (cqlType instanceof MapType mapType) { + return nullableUnion(SchemaBuilder.map().values(toAvroSchema(mapType.getValueType(), udtSchemaCache))); + } + + return nullableUnion(toPrimitiveAvroSchema(cqlType)); + } + + private static Schema toUdtRecordSchema(final UserDefinedType udtType, final Map udtSchemaCache) { + final String namespace = udtType.getKeyspace() == null ? "cassandra" : udtType.getKeyspace().asInternal(); + final String schemaKey = namespace + "." + udtType.getName().asInternal(); + + final Schema cached = udtSchemaCache.get(schemaKey); + if (cached != null) { + return cached; + } + + SchemaBuilder.FieldAssembler fields = SchemaBuilder.record(udtType.getName().asInternal()) + .namespace(namespace) + .fields(); + + final List fieldNames = udtType.getFieldNames(); + final List fieldTypes = udtType.getFieldTypes(); + + for (int i = 0; i < fieldNames.size(); i++) { + fields = fields.name(fieldNames.get(i).asInternal()).type(toAvroSchema(fieldTypes.get(i), udtSchemaCache)).noDefault(); + } + + final Schema schema = fields.endRecord(); + udtSchemaCache.put(schemaKey, schema); + return schema; + } + + private static Schema nullableUnion(final Schema schema) { + return SchemaBuilder.builder().unionOf().nullBuilder().endNull().and().type(schema).endUnion(); + } + + private static Schema toPrimitiveAvroSchema(final DataType cqlType) { + final SchemaBuilder.TypeBuilder typeBuilder = SchemaBuilder.builder(); + + if (cqlType.equals(ASCII) || cqlType.equals(DataTypes.TEXT)) { + return typeBuilder.stringType(); + } + + // Declared with the Avro `uuid` logical type, so AvroTypeUtil#determineDataType resolves the record + // field to the native UUID type rather than STRING - matching the java.util.UUID the driver's own + // codec decodes the column into, with no value-side conversion needed on either side. + if (cqlType.equals(DataTypes.UUID) || cqlType.equals(DataTypes.TIMEUUID)) { + final Schema schema = Schema.create(Schema.Type.STRING); + LogicalTypes.uuid().addToSchema(schema); + return schema; + } + + // TIMESTAMP/DATE/TIME, DECIMAL/VARINT and INET are all declared as a plain Avro string, and the + // driver's own decoded value (Instant/LocalDate/LocalTime, BigDecimal/BigInteger, InetAddress) is + // placed into the record unchanged. The declared type and the runtime class differ, but that is a + // supported combination rather than a defect: RecordFieldType.STRING is a *widening* type over + // DATE/TIME/TIMESTAMP/DECIMAL/BIGINT among others, so DataTypeUtils accepts each of these values for + // a string field and coerces them losslessly (a nanosecond LocalTime renders in full, for instance) + // if and when a downstream writer asks for a string. + // + // Declaring the native NiFi types instead is not an option here: + // - TIMESTAMP/DATE/TIME are backed by java.sql.*, not java.time.*, so pairing them with the + // driver's values makes DataTypeUtils#convertType throw for TIMESTAMP and silently truncate a + // nanosecond TIME to whole seconds. Converting the values to java.sql.* to suit instead loses + // that same sub-second precision, since CQL `time` is nanosecond-resolution and java.sql.Time is + // not. + // - DECIMAL would need Avro's fixed precision/scale, which CQL's arbitrary-precision `decimal` + // cannot supply up front, and a native BIGINT is unreachable through the Avro round trip at all + // (AvroTypeUtil maps record BIGINT to Avro string, which resolves back to record STRING). + // - INET has no native NiFi record type in the first place. + if (cqlType.equals(DataTypes.TIMESTAMP) + || cqlType.equals(DataTypes.DATE) + || cqlType.equals(DataTypes.TIME) + || cqlType.equals(DataTypes.INET) + || cqlType.equals(DataTypes.VARINT) + || cqlType.equals(DataTypes.DECIMAL)) { + return typeBuilder.stringType(); + } + + if (cqlType.equals(DataTypes.BOOLEAN)) { + return typeBuilder.booleanType(); + } + + if (cqlType.equals(DataTypes.INT) + // Avro has no byte/short primitive, so these widen to Avro's int + || cqlType.equals(DataTypes.TINYINT) + || cqlType.equals(DataTypes.SMALLINT)) { + return typeBuilder.intType(); + } + + if (cqlType.equals(DataTypes.BIGINT) || cqlType.equals(DataTypes.COUNTER)) { + return typeBuilder.longType(); + } + + if (cqlType.equals(DataTypes.FLOAT)) { + return typeBuilder.floatType(); + } + + if (cqlType.equals(DataTypes.DOUBLE)) { + return typeBuilder.doubleType(); + } + + if (cqlType.equals(DataTypes.BLOB)) { + return typeBuilder.bytesType(); + } + + // A CQL type with no case above - DURATION, TUPLE, VECTOR, and any future or custom type - is declared + // as STRING rather than failing schema generation. This used to throw IllegalArgumentException here, + // which failed the entire query over one exotic column; CassandraCQLExecutionService#toRecordValue + // converts the value to match via toString(), the same fallback DECIMAL/VARINT/INET already use. + return typeBuilder.stringType(); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/CharacterCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/CharacterCodec.java new file mode 100644 index 000000000000..d35019eebf2a --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/CharacterCodec.java @@ -0,0 +1,70 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; + +import java.nio.ByteBuffer; + +/** + * A CQL {@code text} codec that binds a {@link Character} (what {@code DataTypeUtils} produces for a + * CHAR-typed record field) as a one-character string, since Cassandra has no single-character type. + */ +public class CharacterCodec implements TypeCodec { + private final TypeCodec stringCodec; + + public CharacterCodec() { + this.stringCodec = TypeCodecs.TEXT; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Character.class); + } + + @Override + public DataType getCqlType() { + return stringCodec.getCqlType(); // maps to CQL `text` + } + + @Override + public ByteBuffer encode(Character value, ProtocolVersion protocolVersion) { + return value == null ? null : stringCodec.encode(value.toString(), protocolVersion); + } + + @Override + public Character decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + final String string = stringCodec.decode(bytes, protocolVersion); + return (string == null || string.isEmpty()) ? null : string.charAt(0); + } + + @Override + public String format(Character value) { + return value == null ? "NULL" : stringCodec.format(value.toString()); + } + + @Override + public Character parse(String value) { + final String string = stringCodec.parse(value); + return (string == null || string.isEmpty()) ? null : string.charAt(0); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleBigIntCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleBigIntCodec.java new file mode 100644 index 000000000000..83e4e4f3b824 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleBigIntCodec.java @@ -0,0 +1,70 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; +import org.apache.nifi.serialization.record.util.DataTypeUtils; + +import java.nio.ByteBuffer; + +/** + * A fallback CQL {@code bigint} codec that accepts any long-compatible value (a numeric String, a + * differently-sized {@link Number}), not just an exact {@link Long}. Declared against {@link Object} so + * the registry treats it as a fallback behind the built-in codec rather than rejecting it as a duplicate. + */ +public class FlexibleBigIntCodec implements TypeCodec { + private final TypeCodec bigIntCodec; + + public FlexibleBigIntCodec() { + this.bigIntCodec = TypeCodecs.BIGINT; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Object.class); + } + + @Override + public DataType getCqlType() { + return bigIntCodec.getCqlType(); // maps to CQL `bigint` + } + + @Override + public ByteBuffer encode(Object value, ProtocolVersion protocolVersion) { + return value == null ? null : bigIntCodec.encode(DataTypeUtils.toLong(value, getCqlType().toString()), protocolVersion); + } + + @Override + public Object decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + return bigIntCodec.decode(bytes, protocolVersion); + } + + @Override + public String format(Object value) { + return value == null ? "NULL" : bigIntCodec.format(DataTypeUtils.toLong(value, getCqlType().toString())); + } + + @Override + public Object parse(String value) { + return bigIntCodec.parse(value); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleBooleanCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleBooleanCodec.java new file mode 100644 index 000000000000..f6b323f484a3 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleBooleanCodec.java @@ -0,0 +1,71 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; +import org.apache.nifi.serialization.record.util.DataTypeUtils; + +import java.nio.ByteBuffer; + +/** + * A fallback CQL {@code boolean} codec that accepts any boolean-compatible value (such as a + * {@code "true"}/{@code "false"} String), not just an exact {@link Boolean}. Declared against + * {@link Object} so the registry treats it as a fallback behind the built-in codec rather than + * rejecting it as a duplicate. + */ +public class FlexibleBooleanCodec implements TypeCodec { + private final TypeCodec booleanCodec; + + public FlexibleBooleanCodec() { + this.booleanCodec = TypeCodecs.BOOLEAN; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Object.class); + } + + @Override + public DataType getCqlType() { + return booleanCodec.getCqlType(); // maps to CQL `boolean` + } + + @Override + public ByteBuffer encode(Object value, ProtocolVersion protocolVersion) { + return value == null ? null : booleanCodec.encode(DataTypeUtils.toBoolean(value, getCqlType().toString()), protocolVersion); + } + + @Override + public Object decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + return booleanCodec.decode(bytes, protocolVersion); + } + + @Override + public String format(Object value) { + return value == null ? "NULL" : booleanCodec.format(DataTypeUtils.toBoolean(value, getCqlType().toString())); + } + + @Override + public Object parse(String value) { + return booleanCodec.parse(value); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleCounterCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleCounterCodec.java new file mode 100644 index 000000000000..440b88c81916 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleCounterCodec.java @@ -0,0 +1,71 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.DataTypes; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; + +import java.nio.ByteBuffer; + +/** + * A CQL {@code counter} codec that binds any {@link Number} by narrowing it to a {@code long} and + * delegating to the built-in {@code bigint} codec, since a counter is stored as a 64-bit integer. + */ +public class FlexibleCounterCodec implements TypeCodec { + + private final TypeCodec inner = TypeCodecs.BIGINT; + + @Override + public GenericType getJavaType() { + return GenericType.of(Number.class); + } + + @Override + public DataType getCqlType() { + return DataTypes.COUNTER; + } + + @Override + public ByteBuffer encode(Number value, ProtocolVersion protocolVersion) { + if (value == null) { + return null; + } + return inner.encode(value.longValue(), protocolVersion); + } + + @Override + public Number decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + Long val = inner.decode(bytes, protocolVersion); + return val; // returns as Long, but you can cast if needed + } + + @Override + public String format(Number value) { + return inner.format(value == null ? null : value.longValue()); + } + + @Override + public Number parse(String value) { + return inner.parse(value); + } +} + diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleDateCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleDateCodec.java new file mode 100644 index 000000000000..494361cf6f0d --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleDateCodec.java @@ -0,0 +1,83 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; +import org.apache.nifi.serialization.record.RecordFieldType; +import org.apache.nifi.serialization.record.util.DataTypeUtils; + +import java.nio.ByteBuffer; +import java.sql.Date; +import java.time.LocalDate; +import java.util.Optional; + +/** + * A fallback CQL {@code date} codec that accepts any date-compatible value {@code DataTypeUtils} recognizes, + * including a {@link String} parsed with the fixed ISO {@code yyyy-MM-dd} pattern (what a RecordPath + * {@code format()} primary key override produces). Declared against {@link Object} so the registry treats it + * as a fallback behind the built-in {@link LocalDate} codec rather than rejecting it as a duplicate. + */ +public class FlexibleDateCodec implements TypeCodec { + private static final String DATE_PATTERN = "yyyy-MM-dd"; + + private final TypeCodec dateCodec; + + public FlexibleDateCodec() { + this.dateCodec = TypeCodecs.DATE; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Object.class); + } + + @Override + public DataType getCqlType() { + return dateCodec.getCqlType(); // maps to CQL `date` + } + + @Override + public ByteBuffer encode(Object value, ProtocolVersion protocolVersion) { + return value == null ? null : dateCodec.encode(toLocalDate(value), protocolVersion); + } + + @Override + public Object decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + return dateCodec.decode(bytes, protocolVersion); + } + + @Override + public String format(Object value) { + return value == null ? "NULL" : dateCodec.format(toLocalDate(value)); + } + + @Override + public Object parse(String value) { + return dateCodec.parse(value); + } + + private static LocalDate toLocalDate(final Object value) { + final Object converted = DataTypeUtils.convertType(value, RecordFieldType.DATE.getDataType(), + Optional.of(DATE_PATTERN), Optional.empty(), Optional.empty(), "value"); + return ((Date) converted).toLocalDate(); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleDoubleCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleDoubleCodec.java new file mode 100644 index 000000000000..c7e8cf1cd42e --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleDoubleCodec.java @@ -0,0 +1,70 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; +import org.apache.nifi.serialization.record.util.DataTypeUtils; + +import java.nio.ByteBuffer; + +/** + * A fallback CQL {@code double} codec that accepts any double-compatible value (a numeric String, a + * differently-sized {@link Number}), not just an exact {@link Double}. Declared against {@link Object} so + * the registry treats it as a fallback behind the built-in codec rather than rejecting it as a duplicate. + */ +public class FlexibleDoubleCodec implements TypeCodec { + private final TypeCodec doubleCodec; + + public FlexibleDoubleCodec() { + this.doubleCodec = TypeCodecs.DOUBLE; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Object.class); + } + + @Override + public DataType getCqlType() { + return doubleCodec.getCqlType(); // maps to CQL `double` + } + + @Override + public ByteBuffer encode(Object value, ProtocolVersion protocolVersion) { + return value == null ? null : doubleCodec.encode(DataTypeUtils.toDouble(value, getCqlType().toString()), protocolVersion); + } + + @Override + public Object decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + return doubleCodec.decode(bytes, protocolVersion); + } + + @Override + public String format(Object value) { + return value == null ? "NULL" : doubleCodec.format(DataTypeUtils.toDouble(value, getCqlType().toString())); + } + + @Override + public Object parse(String value) { + return doubleCodec.parse(value); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleFloatCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleFloatCodec.java new file mode 100644 index 000000000000..e793337c8395 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleFloatCodec.java @@ -0,0 +1,70 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; +import org.apache.nifi.serialization.record.util.DataTypeUtils; + +import java.nio.ByteBuffer; + +/** + * A fallback CQL {@code float} codec that accepts any float-compatible value (a numeric String, a + * differently-sized {@link Number}), not just an exact {@link Float}. Declared against {@link Object} so + * the registry treats it as a fallback behind the built-in codec rather than rejecting it as a duplicate. + */ +public class FlexibleFloatCodec implements TypeCodec { + private final TypeCodec floatCodec; + + public FlexibleFloatCodec() { + this.floatCodec = TypeCodecs.FLOAT; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Object.class); + } + + @Override + public DataType getCqlType() { + return floatCodec.getCqlType(); // maps to CQL `float` + } + + @Override + public ByteBuffer encode(Object value, ProtocolVersion protocolVersion) { + return value == null ? null : floatCodec.encode(DataTypeUtils.toFloat(value, getCqlType().toString()), protocolVersion); + } + + @Override + public Object decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + return floatCodec.decode(bytes, protocolVersion); + } + + @Override + public String format(Object value) { + return value == null ? "NULL" : floatCodec.format(DataTypeUtils.toFloat(value, getCqlType().toString())); + } + + @Override + public Object parse(String value) { + return floatCodec.parse(value); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleIntCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleIntCodec.java new file mode 100644 index 000000000000..e77667b40244 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleIntCodec.java @@ -0,0 +1,70 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; +import org.apache.nifi.serialization.record.util.DataTypeUtils; + +import java.nio.ByteBuffer; + +/** + * A fallback CQL {@code int} codec that accepts any int-compatible value (a numeric String, a + * differently-sized {@link Number}), not just an exact {@link Integer}. Declared against {@link Object} so + * the registry treats it as a fallback behind the built-in codec rather than rejecting it as a duplicate. + */ +public class FlexibleIntCodec implements TypeCodec { + private final TypeCodec intCodec; + + public FlexibleIntCodec() { + this.intCodec = TypeCodecs.INT; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Object.class); + } + + @Override + public DataType getCqlType() { + return intCodec.getCqlType(); // maps to CQL `int` + } + + @Override + public ByteBuffer encode(Object value, ProtocolVersion protocolVersion) { + return value == null ? null : intCodec.encode(DataTypeUtils.toInteger(value, getCqlType().toString()), protocolVersion); + } + + @Override + public Object decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + return intCodec.decode(bytes, protocolVersion); + } + + @Override + public String format(Object value) { + return value == null ? "NULL" : intCodec.format(DataTypeUtils.toInteger(value, getCqlType().toString())); + } + + @Override + public Object parse(String value) { + return intCodec.parse(value); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleSmallIntCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleSmallIntCodec.java new file mode 100644 index 000000000000..76d7ffcde9e7 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleSmallIntCodec.java @@ -0,0 +1,70 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; +import org.apache.nifi.serialization.record.util.DataTypeUtils; + +import java.nio.ByteBuffer; + +/** + * A fallback CQL {@code smallint} codec that accepts any short-compatible value (a numeric String, a + * differently-sized {@link Number}), not just an exact {@link Short}. Declared against {@link Object} so + * the registry treats it as a fallback behind the built-in codec rather than rejecting it as a duplicate. + */ +public class FlexibleSmallIntCodec implements TypeCodec { + private final TypeCodec smallIntCodec; + + public FlexibleSmallIntCodec() { + this.smallIntCodec = TypeCodecs.SMALLINT; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Object.class); + } + + @Override + public DataType getCqlType() { + return smallIntCodec.getCqlType(); // maps to CQL `smallint` + } + + @Override + public ByteBuffer encode(Object value, ProtocolVersion protocolVersion) { + return value == null ? null : smallIntCodec.encode(DataTypeUtils.toShort(value, getCqlType().toString()), protocolVersion); + } + + @Override + public Object decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + return smallIntCodec.decode(bytes, protocolVersion); + } + + @Override + public String format(Object value) { + return value == null ? "NULL" : smallIntCodec.format(DataTypeUtils.toShort(value, getCqlType().toString())); + } + + @Override + public Object parse(String value) { + return smallIntCodec.parse(value); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleTimeCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleTimeCodec.java new file mode 100644 index 000000000000..ee3e6ac0bc79 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleTimeCodec.java @@ -0,0 +1,83 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; +import org.apache.nifi.serialization.record.RecordFieldType; +import org.apache.nifi.serialization.record.util.DataTypeUtils; + +import java.nio.ByteBuffer; +import java.sql.Time; +import java.time.LocalTime; +import java.util.Optional; + +/** + * A fallback CQL {@code time} codec that accepts any time-compatible value {@code DataTypeUtils} recognizes, + * including a {@link String} parsed with the fixed {@code HH:mm:ss} pattern (what a RecordPath {@code format()} + * primary key override produces). Declared against {@link Object} so the registry treats it as a fallback + * behind the built-in {@link LocalTime} codec rather than rejecting it as a duplicate. + */ +public class FlexibleTimeCodec implements TypeCodec { + private static final String TIME_PATTERN = "HH:mm:ss"; + + private final TypeCodec timeCodec; + + public FlexibleTimeCodec() { + this.timeCodec = TypeCodecs.TIME; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Object.class); + } + + @Override + public DataType getCqlType() { + return timeCodec.getCqlType(); // maps to CQL `time` + } + + @Override + public ByteBuffer encode(Object value, ProtocolVersion protocolVersion) { + return value == null ? null : timeCodec.encode(toLocalTime(value), protocolVersion); + } + + @Override + public Object decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + return timeCodec.decode(bytes, protocolVersion); + } + + @Override + public String format(Object value) { + return value == null ? "NULL" : timeCodec.format(toLocalTime(value)); + } + + @Override + public Object parse(String value) { + return timeCodec.parse(value); + } + + private static LocalTime toLocalTime(final Object value) { + final Object converted = DataTypeUtils.convertType(value, RecordFieldType.TIME.getDataType(), + Optional.empty(), Optional.of(TIME_PATTERN), Optional.empty(), "value"); + return ((Time) converted).toLocalTime(); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleTinyIntCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleTinyIntCodec.java new file mode 100644 index 000000000000..2b15f1826de4 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/FlexibleTinyIntCodec.java @@ -0,0 +1,70 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; +import org.apache.nifi.serialization.record.util.DataTypeUtils; + +import java.nio.ByteBuffer; + +/** + * A fallback CQL {@code tinyint} codec that accepts any byte-compatible value (a numeric String, a + * differently-sized {@link Number}), not just an exact {@link Byte}. Declared against {@link Object} so + * the registry treats it as a fallback behind the built-in codec rather than rejecting it as a duplicate. + */ +public class FlexibleTinyIntCodec implements TypeCodec { + private final TypeCodec tinyIntCodec; + + public FlexibleTinyIntCodec() { + this.tinyIntCodec = TypeCodecs.TINYINT; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Object.class); + } + + @Override + public DataType getCqlType() { + return tinyIntCodec.getCqlType(); // maps to CQL `tinyint` + } + + @Override + public ByteBuffer encode(Object value, ProtocolVersion protocolVersion) { + return value == null ? null : tinyIntCodec.encode(DataTypeUtils.toByte(value, getCqlType().toString()), protocolVersion); + } + + @Override + public Object decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + return tinyIntCodec.decode(bytes, protocolVersion); + } + + @Override + public String format(Object value) { + return value == null ? "NULL" : tinyIntCodec.format(DataTypeUtils.toByte(value, getCqlType().toString())); + } + + @Override + public Object parse(String value) { + return tinyIntCodec.parse(value); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/JavaSQLDateCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/JavaSQLDateCodec.java new file mode 100644 index 000000000000..d13a4b7b2a72 --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/JavaSQLDateCodec.java @@ -0,0 +1,83 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; + +import java.nio.ByteBuffer; +import java.sql.Date; +import java.time.Instant; +import java.time.LocalDate; +import java.time.ZoneId; + +/** + * A CQL {@code date} codec that binds {@code java.sql.Date} (what {@code DataTypeUtils} produces for a + * DATE-typed record field) by converting to and from {@link LocalDate} in the system default time zone. + */ +public class JavaSQLDateCodec implements TypeCodec { + private final TypeCodec localDateCodec; + + public JavaSQLDateCodec() { + this.localDateCodec = TypeCodecs.DATE; + } + + @Override + public GenericType getJavaType() { + return GenericType.of(Date.class); + } + + @Override + public DataType getCqlType() { + return localDateCodec.getCqlType(); // maps to CQL `date` + } + + @Override + public ByteBuffer encode(Date value, ProtocolVersion protocolVersion) { + return value == null ? null : localDateCodec.encode(toLocalDate(value), protocolVersion); + } + + @Override + public Date decode(ByteBuffer bytes, ProtocolVersion protocolVersion) { + final LocalDate localDate = localDateCodec.decode(bytes, protocolVersion); + return localDate == null ? null : toSqlDate(localDate); + } + + @Override + public String format(Date value) { + return value == null ? "NULL" : localDateCodec.format(toLocalDate(value)); + } + + @Override + public Date parse(String value) { + final LocalDate localDate = localDateCodec.parse(value); + return localDate == null ? null : toSqlDate(localDate); + } + + private static LocalDate toLocalDate(final Date value) { + return LocalDate.ofInstant(Instant.ofEpochMilli(value.getTime()), ZoneId.systemDefault()); + } + + private static Date toSqlDate(final LocalDate localDate) { + final long epochMillis = localDate.atStartOfDay(ZoneId.systemDefault()).toInstant().toEpochMilli(); + return new Date(epochMillis); + } +} diff --git a/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/JavaSQLTimeCodec.java b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/JavaSQLTimeCodec.java new file mode 100644 index 000000000000..62b47ec204ff --- /dev/null +++ b/nifi-extension-bundles/nifi-cql-bundle/nifi-cassandra-session-provider-service/src/main/java/org/apache/nifi/service/cassandra/mapping/JavaSQLTimeCodec.java @@ -0,0 +1,80 @@ +/* + * 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.service.cassandra.mapping; + +import com.datastax.oss.driver.api.core.ProtocolVersion; +import com.datastax.oss.driver.api.core.type.DataType; +import com.datastax.oss.driver.api.core.type.codec.TypeCodec; +import com.datastax.oss.driver.api.core.type.codec.TypeCodecs; +import com.datastax.oss.driver.api.core.type.reflect.GenericType; + +import java.nio.ByteBuffer; +import java.sql.Time; +import java.time.LocalDate; +import java.time.LocalTime; +import java.time.ZoneId; +import java.time.ZonedDateTime; + +/** + * A CQL {@code time} codec that binds {@code java.sql.Time} (what {@code DataTypeUtils} produces for a + * TIME-typed record field) by converting to and from {@link LocalTime} in the system default time zone. + */ +public class JavaSQLTimeCodec implements TypeCodec