gnodet commented on code in PR #25102: URL: https://github.com/apache/camel/pull/25102#discussion_r3647499505
########## components/camel-duckdb/src/main/java/org/apache/camel/component/duckdb/DuckDbException.java: ########## @@ -0,0 +1,28 @@ +/* + * 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.camel.component.duckdb; + +public class DuckDbException extends Exception { Review Comment: **Compilation error:** `DuckDbException` extends `Exception` (checked), but `buildCopySql()` and `resolveFormat()` in `DuckDbProducer` throw it without declaring `throws DuckDbException` in their method signatures. This will cause `javac` to reject the code. Suggested fix — change to `RuntimeException` (more consistent with typical Camel exception patterns): ```suggestion public class DuckDbException extends RuntimeException { ``` ########## components/camel-duckdb/src/main/java/org/apache/camel/component/duckdb/DuckDbProducer.java: ########## @@ -0,0 +1,235 @@ +/* + * 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.camel.component.duckdb; + +import java.io.File; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.List; +import java.util.Locale; +import java.util.Map; + +import org.apache.camel.Exchange; +import org.apache.camel.Message; +import org.apache.camel.WrappedFile; +import org.apache.camel.support.DefaultProducer; +import org.apache.camel.util.ObjectHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DuckDbProducer extends DefaultProducer { + + private static final Logger LOG = LoggerFactory.getLogger(DuckDbProducer.class); + + private final DuckDbEndpoint endpoint; + + public DuckDbProducer(DuckDbEndpoint endpoint) { + super(endpoint); + this.endpoint = endpoint; + } + + @Override + public void process(Exchange exchange) throws Exception { + DuckDbOperation operation = resolveOperation(exchange); + try (ConnectionScope scope = ConnectionScope.open(endpoint)) { + Connection connection = scope.connection(); + switch (operation) { + case EXECUTE -> doExecute(exchange, connection); + case QUERY -> doQuery(exchange, connection); + case INSERT -> doInsert(exchange, connection); + case COPY -> doCopy(exchange, connection); + case PING -> doPing(exchange, connection); + default -> throw new DuckDbException("Unsupported operation: " + operation); + } + } + } + + private void doExecute(Exchange exchange, Connection connection) throws Exception { + String sql = resolveSql(exchange); + if (ObjectHelper.isEmpty(sql)) { + throw new DuckDbException("SQL is required for execute (message body or query option)"); + } + long updated; + try (Statement statement = connection.createStatement()) { + updated = statement.executeUpdate(sql); + } + setRowsWritten(exchange, updated); + exchange.getMessage().setBody(updated); + } + + private void doQuery(Exchange exchange, Connection connection) throws Exception { + String sql = resolveSql(exchange); + if (ObjectHelper.isEmpty(sql)) { + throw new DuckDbException("SQL is required for query (message body, query option or CamelDuckDbQuery header)"); + } + try (Statement statement = connection.createStatement(); + ResultSet resultSet = statement.executeQuery(sql)) { + List<Map<String, Object>> rows = DuckDbResultSetMapper.toListMap(resultSet); + Message message = exchange.getMessage(); + message.setHeader(DuckDbConstants.ROWS_READ, (long) rows.size()); + if (endpoint.getResultFormat() == DuckDbResultFormat.JSON) { + message.setBody(DuckDbResultSetMapper.toJson(rows)); + } else { + message.setBody(rows); + } + } + } + + private void doInsert(Exchange exchange, Connection connection) throws Exception { + String table = resolveTable(exchange); + if (ObjectHelper.isEmpty(table)) { + throw new DuckDbException( + "A table is required for insert. Provide databasePath/table, the table option, or the " + + DuckDbConstants.TABLE + " header."); + } + List<?> body = exchange.getIn().getBody(List.class); + if (body == null) { + throw new DuckDbException("Insert operation expects the message body to be a List of rows"); + } + long written = DuckDbInsertSupport.insertMaps(connection, table, body, endpoint.getBatchSize()); + LOG.debug("Inserted {} rows into {}", written, table); + setRowsWritten(exchange, written); + } + + private void doCopy(Exchange exchange, Connection connection) throws Exception { + String table = resolveTable(exchange); + if (ObjectHelper.isEmpty(table)) { + throw new DuckDbException("A table is required for copy operations"); + } + String path = resolveFilePath(exchange); + String format = resolveFormat(path); + String sql = buildCopySql(table, path, format); + long updated; + try (Statement statement = connection.createStatement()) { + updated = statement.executeUpdate(sql); + } + setRowsWritten(exchange, updated); + } + + private void doPing(Exchange exchange, Connection connection) throws Exception { + boolean ok; + try (Statement statement = connection.createStatement(); + ResultSet rs = statement.executeQuery("SELECT 1")) { + ok = rs.next() && rs.getInt(1) == 1; + } + Message message = exchange.getMessage(); + message.setHeader(DuckDbConstants.PING_OK, ok); + message.setBody(ok); + } + + private String buildCopySql(String table, String path, String format) { Review Comment: This method throws checked `DuckDbException` in the `default` branch but the signature does not declare `throws`. This is part of the compilation error described on `DuckDbException.java` — if `DuckDbException` is changed to extend `RuntimeException`, this is resolved automatically. ########## components/camel-duckdb/src/main/java/org/apache/camel/component/duckdb/DuckDbInsertSupport.java: ########## @@ -0,0 +1,84 @@ +/* + * 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.camel.component.duckdb; + +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +final class DuckDbInsertSupport { + + private DuckDbInsertSupport() { + } + + static long insertMaps(Connection connection, String table, List<?> rows, int batchSize) throws Exception { + if (rows.isEmpty()) { + return 0; + } + if (!(rows.get(0) instanceof Map)) { + throw new DuckDbException("Insert body must be a List of Map entries when using operation=insert"); + } + @SuppressWarnings("unchecked") + List<Map<String, Object>> maps = (List<Map<String, Object>>) rows; + List<String> columns = new ArrayList<>(maps.get(0).keySet()); + if (columns.isEmpty()) { + throw new DuckDbException("Insert row maps must contain at least one column"); + } + + String sql = buildInsertSql(table, columns); + long written = 0; + int effectiveBatch = batchSize > 0 ? batchSize : maps.size(); + try (PreparedStatement ps = connection.prepareStatement(sql)) { + for (int i = 0; i < maps.size(); i++) { + Map<String, Object> row = maps.get(i); + for (int c = 0; c < columns.size(); c++) { + ps.setObject(c + 1, row.get(columns.get(c))); + } + ps.addBatch(); + if ((i + 1) % effectiveBatch == 0 || i + 1 == maps.size()) { + int[] counts = ps.executeBatch(); + for (int count : counts) { + if (count >= 0) { + written += count; + } else { + written += 1; + } + } + } + } + } + return written; + } + + private static String buildInsertSql(String table, List<String> columns) { + StringBuilder sql = new StringBuilder("INSERT INTO "); + sql.append(table).append(" ("); Review Comment: **Defense-in-depth:** Table and column names are interpolated directly into SQL without quoting. While route authors are trusted per Camel's security model, wrapping identifiers in double quotes provides defense against reserved words and special characters in identifiers: ```java sql.append('"').append(table).append('"').append(" ("); ``` Same applies to column names in this method and in `DuckDbProducer.buildCopySql()`. ########## components/camel-duckdb/src/main/java/org/apache/camel/component/duckdb/DuckDbEndpoint.java: ########## @@ -0,0 +1,214 @@ +/* + * 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.camel.component.duckdb; + +import java.sql.Connection; +import java.sql.DriverManager; + +import javax.sql.DataSource; + +import org.apache.camel.Category; +import org.apache.camel.Consumer; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.UriEndpoint; +import org.apache.camel.spi.UriParam; +import org.apache.camel.spi.UriPath; +import org.apache.camel.support.DefaultEndpoint; +import org.apache.camel.util.ObjectHelper; + +/** + * Interact with <a href="https://duckdb.org/";>DuckDB</a>, the in-process analytical SQL database, for embedded + * analytics workloads. + */ +@UriEndpoint(firstVersion = "4.22.0", scheme = "duckdb", title = "DuckDB", + syntax = "duckdb:databasePath", category = { Category.DATABASE, Category.BIGDATA }, + producerOnly = true, headersClass = DuckDbConstants.class) +public class DuckDbEndpoint extends DefaultEndpoint { + + private DataSource dataSource; + private Connection connection; + private Connection injectedConnection; + private boolean connectionOwned; + + @UriPath(description = "Database path (:memory: or a file path). An optional /table segment sets the default table.") + private String databasePath = ":memory:"; + @UriParam(description = "Target table for insert and copy operations. Can also be set in the path as databasePath/table" + + " or overridden with the CamelDuckDbTable header.") + private String table; + @UriParam(description = "Full JDBC URL override. When set, databasePath is not used to build the URL.") + private String jdbcUrl; + @UriParam(defaultValue = "EXECUTE", description = "The operation to perform: execute, query, insert, copy or ping.") + private DuckDbOperation operation = DuckDbOperation.EXECUTE; + @UriParam(description = "Static SQL for query when the message body is empty.") + private String query; + @UriParam(defaultValue = "0", + description = "Batch size for insert when the body is a List. 0 inserts all rows in one JDBC batch.") + private int batchSize; + @UriParam(defaultValue = "auto", + description = "File format for copy: csv, parquet, json or auto (inferred from the file extension).") + private String format = "auto"; + @UriParam(defaultValue = "false", description = "Open the embedded database file in read-only mode when supported.") + private boolean readOnly; Review Comment: **Dead parameter:** The `readOnly` `@UriParam` is declared and documented, but is never referenced in `doStart()`, `resolveJdbcUrl()`, or anywhere in connection setup. Users setting `readOnly=true` would expect read-only mode but it has no effect. Either implement it (e.g., append `?access_mode=read_only` to the JDBC URL or pass a `Properties` object to `DriverManager.getConnection()`) or remove it entirely. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
