You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2025/10/28 10:30:26 UTC
(camel) branch main updated: [CAMEL-22606] camel-google-bigquery - Support stream list output type for SELECT queries (#19695)
This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 87660950ac4d [CAMEL-22606] camel-google-bigquery - Support stream list output type for SELECT queries (#19695)
87660950ac4d is described below
commit 87660950ac4d3521ec7ebb262f24feb320ed17c9
Author: Volodymyr Vysotskyi <vo...@fiverr.com>
AuthorDate: Tue Oct 28 12:30:20 2025 +0200
[CAMEL-22606] camel-google-bigquery - Support stream list output type for SELECT queries (#19695)
* CAMEL-22606: camel-google-bigquery - Support stream list output type for SELECT queries
* CAMEL-22606: camel-google-bigquery - Add integration tests for stream list output type and fix mapping
---
.../catalog/components/google-bigquery-sql.json | 8 +-
.../camel-google/camel-google-bigquery/pom.xml | 5 +
.../sql/GoogleBigQuerySQLEndpointConfigurer.java | 12 ++
.../sql/GoogleBigQuerySQLEndpointUriFactory.java | 4 +-
.../google/bigquery/sql/google-bigquery-sql.json | 8 +-
.../main/docs/google-bigquery-sql-component.adoc | 111 ++++++++++++
.../google/bigquery/sql/FieldValueListMapper.java | 69 ++++++++
.../sql/GoogleBigQuerySQLConfiguration.java | 25 ++-
.../bigquery/sql/GoogleBigQuerySQLProducer.java | 194 ++++++++++++++-------
.../component/google/bigquery/sql/OutputType.java | 39 +++++
.../integration/sql/SqlQuerySelectAndStreamIT.java | 142 +++++++++++++++
.../unit/sql/FieldValueListMapperTest.java | 168 ++++++++++++++++++
.../sql/GoogleBigQuerySQLProducerBaseTest.java | 5 +
.../GoogleBigQuerySQLProducerPaginationTest.java | 177 +++++++++++++++++++
.../sql/GoogleBigQuerySQLProducerSelectTest.java | 17 +-
.../GoogleBigQuerySQLProducerStreamListTest.java | 193 ++++++++++++++++++++
.../GoogleBigQuerySQLProducerWithParamersTest.java | 25 +--
.../GoogleBigQuerySQLProducerWithPatternTest.java | 3 +
.../src/test/resources/sql/select.sql | 33 ++++
...StreamListIterator.java => JdbcBeanMapper.java} | 35 +---
.../apache/camel/component/jdbc/JdbcProducer.java | 9 +-
.../java/org/apache/camel/support/RowMapper.java | 37 ++++
.../apache/camel/support/StreamListIterator.java | 53 ++++++
.../GoogleBigQuerySQLEndpointBuilderFactory.java | 62 +++++++
24 files changed, 1298 insertions(+), 136 deletions(-)
diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/google-bigquery-sql.json b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/google-bigquery-sql.json
index 4958acd4673b..45c460f262ab 100644
--- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/google-bigquery-sql.json
+++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/google-bigquery-sql.json
@@ -39,8 +39,10 @@
"projectId": { "index": 0, "kind": "path", "displayName": "Project Id", "group": "common", "label": "common", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Google Cloud Project Id" },
"queryString": { "index": 1, "kind": "path", "displayName": "Query String", "group": "common", "label": "common", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "supportFileReference": true, "largeInput": true, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "BigQuery standard [...]
"connectionFactory": { "index": 2, "kind": "parameter", "displayName": "Connection Factory", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.google.bigquery.GoogleBigQueryConnectionFactory", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description [...]
- "pageToken": { "index": 3, "kind": "parameter", "displayName": "Page Token", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Page token, returned by a previous call, to request the next page of results" },
- "lazyStartProducer": { "index": 4, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...]
- "serviceAccountKey": { "index": 5, "kind": "parameter", "displayName": "Service Account Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Service account key in json format to authen [...]
+ "outputType": { "index": 3, "kind": "parameter", "displayName": "Output Type", "group": "producer", "label": "producer", "required": false, "type": "enum", "javaType": "org.apache.camel.component.google.bigquery.sql.OutputType", "enum": [ "SELECT_LIST", "STREAM_LIST" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "SELECT_LIST", "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "co [...]
+ "pageSize": { "index": 4, "kind": "parameter", "displayName": "Page Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Size of a page to request. Unset by default" },
+ "pageToken": { "index": 5, "kind": "parameter", "displayName": "Page Token", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Page token, returned by a previous call, to request the next page of results" },
+ "lazyStartProducer": { "index": 6, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...]
+ "serviceAccountKey": { "index": 7, "kind": "parameter", "displayName": "Service Account Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Service account key in json format to authenticate an application a [...]
}
}
diff --git a/components/camel-google/camel-google-bigquery/pom.xml b/components/camel-google/camel-google-bigquery/pom.xml
index 6e55bf86315e..12e39eca13c6 100644
--- a/components/camel-google/camel-google-bigquery/pom.xml
+++ b/components/camel-google/camel-google-bigquery/pom.xml
@@ -63,5 +63,10 @@
<version>${mockito-version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/components/camel-google/camel-google-bigquery/src/generated/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLEndpointConfigurer.java b/components/camel-google/camel-google-bigquery/src/generated/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLEndpointConfigurer.java
index 81ac06d61abc..22415975ca82 100644
--- a/components/camel-google/camel-google-bigquery/src/generated/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLEndpointConfigurer.java
+++ b/components/camel-google/camel-google-bigquery/src/generated/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLEndpointConfigurer.java
@@ -27,6 +27,10 @@ public class GoogleBigQuerySQLEndpointConfigurer extends PropertyConfigurerSuppo
case "connectionFactory": target.getConfiguration().setConnectionFactory(property(camelContext, org.apache.camel.component.google.bigquery.GoogleBigQueryConnectionFactory.class, value)); return true;
case "lazystartproducer":
case "lazyStartProducer": target.setLazyStartProducer(property(camelContext, boolean.class, value)); return true;
+ case "outputtype":
+ case "outputType": target.getConfiguration().setOutputType(property(camelContext, org.apache.camel.component.google.bigquery.sql.OutputType.class, value)); return true;
+ case "pagesize":
+ case "pageSize": target.getConfiguration().setPageSize(property(camelContext, long.class, value)); return true;
case "pagetoken":
case "pageToken": target.getConfiguration().setPageToken(property(camelContext, java.lang.String.class, value)); return true;
case "serviceaccountkey":
@@ -47,6 +51,10 @@ public class GoogleBigQuerySQLEndpointConfigurer extends PropertyConfigurerSuppo
case "connectionFactory": return org.apache.camel.component.google.bigquery.GoogleBigQueryConnectionFactory.class;
case "lazystartproducer":
case "lazyStartProducer": return boolean.class;
+ case "outputtype":
+ case "outputType": return org.apache.camel.component.google.bigquery.sql.OutputType.class;
+ case "pagesize":
+ case "pageSize": return long.class;
case "pagetoken":
case "pageToken": return java.lang.String.class;
case "serviceaccountkey":
@@ -63,6 +71,10 @@ public class GoogleBigQuerySQLEndpointConfigurer extends PropertyConfigurerSuppo
case "connectionFactory": return target.getConfiguration().getConnectionFactory();
case "lazystartproducer":
case "lazyStartProducer": return target.isLazyStartProducer();
+ case "outputtype":
+ case "outputType": return target.getConfiguration().getOutputType();
+ case "pagesize":
+ case "pageSize": return target.getConfiguration().getPageSize();
case "pagetoken":
case "pageToken": return target.getConfiguration().getPageToken();
case "serviceaccountkey":
diff --git a/components/camel-google/camel-google-bigquery/src/generated/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLEndpointUriFactory.java b/components/camel-google/camel-google-bigquery/src/generated/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLEndpointUriFactory.java
index 8ae41023a623..d1bd966a3c25 100644
--- a/components/camel-google/camel-google-bigquery/src/generated/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLEndpointUriFactory.java
+++ b/components/camel-google/camel-google-bigquery/src/generated/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLEndpointUriFactory.java
@@ -23,9 +23,11 @@ public class GoogleBigQuerySQLEndpointUriFactory extends org.apache.camel.suppor
private static final Set<String> SECRET_PROPERTY_NAMES;
private static final Map<String, String> MULTI_VALUE_PREFIXES;
static {
- Set<String> props = new HashSet<>(6);
+ Set<String> props = new HashSet<>(8);
props.add("connectionFactory");
props.add("lazyStartProducer");
+ props.add("outputType");
+ props.add("pageSize");
props.add("pageToken");
props.add("projectId");
props.add("queryString");
diff --git a/components/camel-google/camel-google-bigquery/src/generated/resources/META-INF/org/apache/camel/component/google/bigquery/sql/google-bigquery-sql.json b/components/camel-google/camel-google-bigquery/src/generated/resources/META-INF/org/apache/camel/component/google/bigquery/sql/google-bigquery-sql.json
index 4958acd4673b..45c460f262ab 100644
--- a/components/camel-google/camel-google-bigquery/src/generated/resources/META-INF/org/apache/camel/component/google/bigquery/sql/google-bigquery-sql.json
+++ b/components/camel-google/camel-google-bigquery/src/generated/resources/META-INF/org/apache/camel/component/google/bigquery/sql/google-bigquery-sql.json
@@ -39,8 +39,10 @@
"projectId": { "index": 0, "kind": "path", "displayName": "Project Id", "group": "common", "label": "common", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Google Cloud Project Id" },
"queryString": { "index": 1, "kind": "path", "displayName": "Query String", "group": "common", "label": "common", "required": true, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "supportFileReference": true, "largeInput": true, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "BigQuery standard [...]
"connectionFactory": { "index": 2, "kind": "parameter", "displayName": "Connection Factory", "group": "producer", "label": "", "required": false, "type": "object", "javaType": "org.apache.camel.component.google.bigquery.GoogleBigQueryConnectionFactory", "deprecated": false, "deprecationNote": "", "autowired": true, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description [...]
- "pageToken": { "index": 3, "kind": "parameter", "displayName": "Page Token", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Page token, returned by a previous call, to request the next page of results" },
- "lazyStartProducer": { "index": 4, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...]
- "serviceAccountKey": { "index": 5, "kind": "parameter", "displayName": "Service Account Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "deprecationNote": "", "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Service account key in json format to authen [...]
+ "outputType": { "index": 3, "kind": "parameter", "displayName": "Output Type", "group": "producer", "label": "producer", "required": false, "type": "enum", "javaType": "org.apache.camel.component.google.bigquery.sql.OutputType", "enum": [ "SELECT_LIST", "STREAM_LIST" ], "deprecated": false, "autowired": false, "secret": false, "defaultValue": "SELECT_LIST", "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "co [...]
+ "pageSize": { "index": 4, "kind": "parameter", "displayName": "Page Size", "group": "producer", "label": "producer", "required": false, "type": "integer", "javaType": "long", "deprecated": false, "autowired": false, "secret": false, "defaultValue": 0, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Size of a page to request. Unset by default" },
+ "pageToken": { "index": 5, "kind": "parameter", "displayName": "Page Token", "group": "producer", "label": "producer", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Page token, returned by a previous call, to request the next page of results" },
+ "lazyStartProducer": { "index": 6, "kind": "parameter", "displayName": "Lazy Start Producer", "group": "producer (advanced)", "label": "producer,advanced", "required": false, "type": "boolean", "javaType": "boolean", "deprecated": false, "autowired": false, "secret": false, "defaultValue": false, "description": "Whether the producer should be started lazy (on the first message). By starting lazy you can use this to allow CamelContext and routes to startup in situations where a produc [...]
+ "serviceAccountKey": { "index": 7, "kind": "parameter", "displayName": "Service Account Key", "group": "security", "label": "security", "required": false, "type": "string", "javaType": "java.lang.String", "deprecated": false, "autowired": false, "secret": false, "configurationClass": "org.apache.camel.component.google.bigquery.sql.GoogleBigQuerySQLConfiguration", "configurationField": "configuration", "description": "Service account key in json format to authenticate an application a [...]
}
}
diff --git a/components/camel-google/camel-google-bigquery/src/main/docs/google-bigquery-sql-component.adoc b/components/camel-google/camel-google-bigquery/src/main/docs/google-bigquery-sql-component.adoc
index 53f860327b12..b67d6a94287b 100644
--- a/components/camel-google/camel-google-bigquery/src/main/docs/google-bigquery-sql-component.adoc
+++ b/components/camel-google/camel-google-bigquery/src/main/docs/google-bigquery-sql-component.adoc
@@ -114,5 +114,116 @@ include::partial$component-endpoint-headers.adoc[]
Google BigQuery SQL endpoint expects the payload to be either empty or a map of query parameters.
+== Query Types
+
+The component supports both SELECT and DML (INSERT, UPDATE, DELETE) queries:
+
+* **SELECT queries** - Return data based on configured `outputType` (SELECT_LIST or STREAM_LIST)
+* **DML queries** - Return the number of affected rows as `Long`
+
+== Output Types
+
+The component supports two output types for SELECT queries, configured using the `outputType` parameter
+
+=== SELECT_LIST (default)
+
+Returns all query results as a `List<Map<String, Object>>` loaded into memory.
+
+*Characteristics:*
+
+* All rows from current page loaded into memory
+* Message body: `List<Map<String, Object>>`
+* Manual pagination via `pageSize` and `pageToken` parameters
+* Sets `CamelGoogleBigQueryNextPageToken` header for next page
+* Sets `CamelGoogleBigQueryJobId` header for continued pagination with same job
+* Results can be accessed multiple times
+
+=== STREAM_LIST
+
+Returns an `Iterator<Map<String, Object>>` for memory-efficient streaming of large result sets. Rows are loaded lazily as the iterator is consumed.
+
+*Characteristics:*
+
+* Message body: `Iterator<Map<String, Object>>`
+* Pagination handled automatically by BigQuery SDK using `pageSize` for internal page fetching
+* Does NOT set `CamelGoogleBigQueryNextPageToken` or `CamelGoogleBigQueryJobId` headers
+* Does NOT support manual `pageToken` control
+* Memory-efficient - only current page in memory
+* Iterator can only be consumed once
+
+*Note:* The `.streaming()` modifier on Split EIP is required to prevent loading all rows into memory before splitting.
+
+*When to use STREAM_LIST:*
+
+* Result sets with millions of rows
+* Memory-constrained environments
+* One-time sequential processing of rows
+* When using Split EIP in streaming mode
+
+== Data Type Mapping
+
+BigQuery result types are mapped to Java types:
+
+[cols="1,1,3"]
+|===
+|BigQuery Type|Java Type|Notes
+
+|INT64
+|Long
+|
+
+|FLOAT64
+|Double
+|
+
+|NUMERIC
+|BigDecimal
+|
+
+|STRING
+|String
+|
+
+|BOOL
+|Boolean
+|
+
+|BYTES
+|byte[]
+|
+
+|DATE
+|String
+|ISO 8601 format
+
+|DATETIME
+|String
+|ISO 8601 format
+
+|TIME
+|String
+|ISO 8601 format
+
+|TIMESTAMP
+|String
+|ISO 8601 format with timezone
+
+|STRUCT (RECORD)
+|Map<String, Object>
+|Recursively converted to nested maps
+
+|ARRAY (REPEATED)
+|List<Object>
+|Elements recursively converted based on their type
+
+|RANGE
+|Map<String, Object>
+|Map with range start and end values
+
+|NULL
+|null
+|
+|===
+
include::spring-boot:partial$starter.adoc[]
diff --git a/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/FieldValueListMapper.java b/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/FieldValueListMapper.java
new file mode 100644
index 000000000000..45cb7f3d301e
--- /dev/null
+++ b/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/FieldValueListMapper.java
@@ -0,0 +1,69 @@
+/*
+ * 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.google.bigquery.sql;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import com.google.cloud.bigquery.*;
+import org.apache.camel.support.RowMapper;
+
+/**
+ * Maps BigQuery {@link FieldValueList} rows to {@code Map<String, Object>}.
+ * <p>
+ * Handles all BigQuery data types:
+ * <ul>
+ * <li>Primitives (INT64, STRING, FLOAT64, BOOL, etc.) → Java equivalents</li>
+ * <li>NULL → null</li>
+ * <li>RECORD (STRUCT) → nested Map (recursive)</li>
+ * <li>REPEATED (ARRAY) → List (recursive)</li>
+ * <li>RANGE → Map with range values</li>
+ * </ul>
+ */
+public class FieldValueListMapper implements RowMapper<FieldValueList, Map<String, Object>> {
+ private final FieldList schema;
+
+ public FieldValueListMapper(FieldList schema) {
+ this.schema = schema;
+ }
+
+ @Override
+ public Map<String, Object> map(FieldValueList row) {
+ return map(schema, row);
+ }
+
+ private Map<String, Object> map(FieldList schema, FieldValueList row) {
+ Map<String, Object> rowMap = new HashMap<>();
+ for (int i = 0; i < schema.size(); i++) {
+ var field = schema.get(i);
+ rowMap.put(field.getName(), convertValue(field, row.get(i)));
+ }
+ return rowMap;
+ }
+
+ private Object convertValue(Field field, FieldValue fieldValue) {
+ return switch (fieldValue.getAttribute()) {
+ case RECORD -> map(field.getSubFields(), fieldValue.getRecordValue());
+ case REPEATED -> fieldValue.getRepeatedValue().stream()
+ .map(element -> convertValue(field, element))
+ .collect(Collectors.toList());
+ case RANGE -> fieldValue.getRangeValue().getValues();
+ case PRIMITIVE -> fieldValue.getValue();
+ };
+ }
+}
diff --git a/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLConfiguration.java b/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLConfiguration.java
index 2cf61e16df4a..deac850258b7 100644
--- a/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLConfiguration.java
+++ b/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLConfiguration.java
@@ -40,12 +40,17 @@ public class GoogleBigQuerySQLConfiguration implements Cloneable {
@UriParam(label = "security",
description = "Service account key in json format to authenticate an application as a service account to google cloud platform")
- @Metadata(required = false)
private String serviceAccountKey;
@UriParam(label = "producer", description = "Page token, returned by a previous call, to request the next page of results")
private String pageToken;
+ @UriParam(label = "producer", description = "Size of a page to request. Unset by default", defaultValue = "0")
+ private long pageSize;
+
+ @UriParam(label = "producer", description = "Output mode type for select queries", defaultValue = "SELECT_LIST")
+ private OutputType outputType = OutputType.SELECT_LIST;
+
public void parseRemaining(String remaining) {
int indexOfColon = remaining.indexOf(':');
@@ -104,6 +109,24 @@ public class GoogleBigQuerySQLConfiguration implements Cloneable {
return this;
}
+ public long getPageSize() {
+ return pageSize;
+ }
+
+ public GoogleBigQuerySQLConfiguration setPageSize(long pageSize) {
+ this.pageSize = pageSize;
+ return this;
+ }
+
+ public OutputType getOutputType() {
+ return outputType;
+ }
+
+ public GoogleBigQuerySQLConfiguration setOutputType(OutputType outputType) {
+ this.outputType = outputType;
+ return this;
+ }
+
public GoogleBigQuerySQLConfiguration copy() {
try {
return (GoogleBigQuerySQLConfiguration) super.clone();
diff --git a/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLProducer.java b/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLProducer.java
index 7f271f990b1c..f8589defc9ef 100644
--- a/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLProducer.java
+++ b/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/GoogleBigQuerySQLProducer.java
@@ -21,14 +21,12 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.UUID;
+import java.util.stream.StreamSupport;
import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.BigQuery.QueryResultsOption;
import com.google.cloud.bigquery.BigQueryException;
-import com.google.cloud.bigquery.Field;
-import com.google.cloud.bigquery.FieldValueList;
import com.google.cloud.bigquery.Job;
-import com.google.cloud.bigquery.JobException;
import com.google.cloud.bigquery.JobId;
import com.google.cloud.bigquery.JobInfo;
import com.google.cloud.bigquery.JobStatistics;
@@ -42,19 +40,21 @@ import org.apache.camel.Message;
import org.apache.camel.RuntimeExchangeException;
import org.apache.camel.component.google.bigquery.GoogleBigQueryConstants;
import org.apache.camel.support.DefaultProducer;
+import org.apache.camel.support.StreamListIterator;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Generic BigQuery Producer
+ * Producer for executing SQL queries against Google BigQuery. Supports both DML (INSERT, UPDATE, DELETE) and SELECT
+ * queries with parameterized queries, pagination, and streaming results.
*/
public class GoogleBigQuerySQLProducer extends DefaultProducer {
private static final Logger LOG = LoggerFactory.getLogger(GoogleBigQuerySQLProducer.class);
private final GoogleBigQuerySQLConfiguration configuration;
- private BigQuery bigquery;
+ private final BigQuery bigquery;
private String query;
private Set<String> queryParameterNames;
@@ -66,14 +66,21 @@ public class GoogleBigQuerySQLProducer extends DefaultProducer {
}
/**
- * Process the exchange The incoming exchange can be a grouped exchange in which case all the exchanges will be
- * combined. The incoming can be
+ * Processes the exchange by executing a SQL query against BigQuery.
+ * <p>
+ * Query parameters can be provided via:
* <ul>
- * <li>A map where all map keys will map to field records. One map object maps to one bigquery row</li>
- * <li>A list of maps. Each entry in the list will map to one bigquery row</li>
+ * <li>Message body as {@code Map<String, Object>}</li>
+ * <li>Message headers</li>
* </ul>
- * The incoming message is expected to be a List of Maps The assumptions: - All incoming records go into the same
- * table - Incoming records sorted by the timestamp
+ * <p>
+ * For SELECT queries, results are returned based on {@link OutputType}:
+ * <ul>
+ * <li>SELECT_LIST - {@code List<Map<String, Object>>} with pagination headers</li>
+ * <li>STREAM_LIST - {@code Iterator<Map<String, Object>>} for streaming</li>
+ * </ul>
+ * <p>
+ * For DML queries, returns the number of affected rows as {@code Long}.
*/
@Override
public void process(Exchange exchange) throws Exception {
@@ -84,32 +91,57 @@ public class GoogleBigQuerySQLProducer extends DefaultProducer {
message.setHeader(GoogleBigQueryConstants.TRANSLATED_QUERY, translatedQuery);
JobId jobId = message.getHeader(GoogleBigQueryConstants.JOB_ID, JobId.class);
- String pageToken = message.getHeader(GoogleBigQueryConstants.PAGE_TOKEN, String.class);
- if (pageToken == null && configuration.getPageToken() != null) {
- pageToken = configuration.getPageToken();
+ Job job = executeJob(jobId, translatedQuery, queryParameters);
+
+ if (isSelectQueryJob(job)) {
+ processSelectQueryJob(message, job);
+ } else {
+ long affectedRows = job.<JobStatistics.QueryStatistics> getStatistics().getNumDmlAffectedRows();
+ LOG.debug("The query {} affected {} rows", query, affectedRows);
+ message.setBody(affectedRows);
}
+ }
+
+ /**
+ * Processes SELECT query results based on the configured output type. Both types use pageSize for fetching results
+ * from BigQuery. For SELECT_LIST, loads current page into memory and sets pagination headers. For STREAM_LIST,
+ * creates an iterator that automatically fetches pages using pageSize.
+ */
+ private void processSelectQueryJob(Message message, Job job) throws Exception {
+ long pageSize = configuration.getPageSize();
+ String pageToken = message.getHeader(GoogleBigQueryConstants.PAGE_TOKEN, configuration::getPageToken, String.class);
- Object queryResult = executeSQL(jobId, translatedQuery, queryParameters, pageToken);
+ TableResult result = getTableResult(job, pageSize, pageToken);
+ Schema schema = result.getSchema();
- if (queryResult instanceof Long) {
- LOG.debug("The query {} affected {} rows", query, queryResult);
- message.setBody(queryResult);
- } else if (queryResult instanceof TableResult result) {
- Schema schema = result.getSchema();
- if (schema != null) {
+ if (schema == null) {
+ LOG.debug("Query result schema is null. Unable to process the result set.");
+ message.setBody(result.getTotalRows());
+ return;
+ }
+
+ switch (configuration.getOutputType()) {
+ case SELECT_LIST -> {
List<Map<String, Object>> rows = processSelectResult(result, schema);
LOG.debug("The query {} returned {} rows", query, rows.size());
message.setBody(rows);
message.setHeader(GoogleBigQueryConstants.NEXT_PAGE_TOKEN, result.getNextPageToken());
- } else {
- LOG.debug("Query result schema is null. Unable to process the result set.");
- message.setBody(result.getTotalRows());
+ message.setHeader(GoogleBigQueryConstants.JOB_ID, job.getJobId());
+ }
+ case STREAM_LIST -> {
+ var iterator = new StreamListIterator<>(
+ new FieldValueListMapper(schema.getFields()),
+ result.iterateAll().iterator());
+ message.setBody(iterator);
}
}
}
- private Object executeSQL(JobId jobId, String translatedQuery, Map<String, Object> queryParameters, String pageToken)
- throws Exception {
+ /**
+ * Executes a BigQuery job, either by retrieving an existing job or creating a new one. If jobId is provided,
+ * retrieves the existing job; otherwise creates a new query job. Waits for the job to complete before returning.
+ */
+ private Job executeJob(JobId jobId, String translatedQuery, Map<String, Object> queryParameters) throws Exception {
QueryJobConfiguration.Builder builder = QueryJobConfiguration.newBuilder(translatedQuery)
.setUseLegacySql(false);
@@ -122,40 +154,71 @@ public class GoogleBigQuerySQLProducer extends DefaultProducer {
LOG.trace("Sending query to bigquery standard sql: {}", translatedQuery);
}
- JobId queryJobId;
- if (ObjectHelper.isNotEmpty(jobId)) {
- queryJobId = jobId;
- } else {
- queryJobId = JobId.of(configuration.getProjectId(), UUID.randomUUID().toString());
- }
+ var job = ObjectHelper.isNotEmpty(jobId)
+ ? bigquery.getJob(jobId)
+ : bigquery.create(getJobInfo(queryJobConfiguration));
- Job job = bigquery.create(JobInfo.of(queryJobId, queryJobConfiguration)).waitFor();
- JobStatistics.QueryStatistics statistics = job.getStatistics();
- TableResult result;
- if (pageToken != null) {
- result = job.getQueryResults(BigQuery.QueryResultsOption.pageToken(pageToken));
- } else {
- result = job.getQueryResults();
- }
- Long numAffectedRows = statistics.getNumDmlAffectedRows();
+ return job.waitFor();
+ } catch (BigQueryException e) {
+ throw new Exception("Query " + translatedQuery + " failed: " + e.getError(), e);
+ }
+ }
- if (LOG.isTraceEnabled()) {
- LOG.trace("Query {} - Affected rows {} - Result {}", translatedQuery, numAffectedRows, result);
- }
+ /**
+ * Creates JobInfo with a random job ID for the given query configuration.
+ */
+ private JobInfo getJobInfo(QueryJobConfiguration queryJobConfiguration) {
+ return JobInfo.of(
+ JobId.newBuilder()
+ .setRandomJob()
+ .setProject(configuration.getProjectId())
+ .build(),
+ queryJobConfiguration);
+ }
- //numAffectedRows is present only for DML statements INSERT, UPDATE or DELETE.
- if (numAffectedRows != null) {
- return numAffectedRows;
- }
- //in other cases (SELECT), process results
- return result;
- } catch (JobException e) {
- throw new Exception("Query " + translatedQuery + " failed: " + e.getErrors(), e);
+ /**
+ * Retrieves query results from a completed job with optional pagination.
+ */
+ private TableResult getTableResult(Job job, long pageSize, String pageToken)
+ throws Exception {
+ String translatedQuery = job.<QueryJobConfiguration> getConfiguration().getQuery();
+ try {
+ QueryResultsOption[] queryResultsOptions = getQueryResultsOptions(pageSize, pageToken);
+ return job.getQueryResults(queryResultsOptions);
} catch (BigQueryException e) {
throw new Exception("Query " + translatedQuery + " failed: " + e.getError(), e);
}
}
+ /**
+ * Builds query result options array from pageSize and pageToken. Only includes options with non-default values
+ * (pageSize > 0, pageToken != null).
+ */
+ private static QueryResultsOption[] getQueryResultsOptions(long pageSize, String pageToken) {
+ List<QueryResultsOption> options = new ArrayList<>();
+ if (pageSize > 0) {
+ options.add(QueryResultsOption.pageSize(pageSize));
+ }
+ if (pageToken != null) {
+ options.add(QueryResultsOption.pageToken(pageToken));
+ }
+ return options.toArray(new QueryResultsOption[0]);
+ }
+
+ /**
+ * Checks if the job is a SELECT query by examining its statement type.
+ */
+ private static boolean isSelectQueryJob(Job job) {
+ JobStatistics.QueryStatistics statistics = job.getStatistics();
+ return statistics.getStatementType().equals(JobStatistics.QueryStatistics.StatementType.SELECT);
+ }
+
+ /**
+ * Extracts query parameters from exchange headers and body. Parameters are identified by names found in the query
+ * (e.g., @paramName). Body values take precedence over header values.
+ *
+ * @throws RuntimeExchangeException if a required parameter is not found
+ */
private Map<String, Object> extractParameters(Exchange exchange) {
if (queryParameterNames == null || queryParameterNames.isEmpty()) {
return null;
@@ -206,22 +269,15 @@ public class GoogleBigQuerySQLProducer extends DefaultProducer {
}
/**
- * Processes a TableResult from a SELECT query, extracting the rows as a List of Maps.
- *
- * @param result the TableResult returned from BigQuery
- * @param schema the Schema returned from BigQuery
- * @return a List of Maps where each Map represents a row of the result
+ * Converts TableResult to a list of maps for SELECT_LIST output type. Each map represents one row with field names
+ * as keys.
*/
private List<Map<String, Object>> processSelectResult(TableResult result, Schema schema) {
- List<Map<String, Object>> rows = new ArrayList<>();
- for (FieldValueList row : result.getValues()) {
- Map<String, Object> rowMap = new HashMap<>();
- for (Field field : schema.getFields()) {
- rowMap.put(field.getName(), row.get(field.getName()).getValue());
- }
- rows.add(rowMap);
- }
- return rows;
+ var mapper = new FieldValueListMapper(schema.getFields());
+
+ return StreamSupport.stream(result.getValues().spliterator(), false)
+ .map(mapper::map)
+ .toList();
}
@Override
@@ -229,6 +285,10 @@ public class GoogleBigQuerySQLProducer extends DefaultProducer {
return (GoogleBigQuerySQLEndpoint) super.getEndpoint();
}
+ /**
+ * Initializes the producer by resolving the query string and extracting parameter names. Query resolution supports
+ * file references and variable substitution.
+ */
@Override
protected void doStart() throws Exception {
super.doStart();
diff --git a/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/OutputType.java b/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/OutputType.java
new file mode 100644
index 000000000000..3135d16a926b
--- /dev/null
+++ b/components/camel-google/camel-google-bigquery/src/main/java/org/apache/camel/component/google/bigquery/sql/OutputType.java
@@ -0,0 +1,39 @@
+/*
+ * 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.google.bigquery.sql;
+
+/**
+ * Output type for SELECT query results.
+ */
+public enum OutputType {
+ /**
+ * Returns all results as {@code List<Map<String, Object>>} (default). Loads all rows into memory. Sets pagination
+ * headers for manual page control.
+ * <p>
+ * Use when: result set fits in memory, need multiple access to rows, require manual pagination.
+ */
+ SELECT_LIST,
+
+ /**
+ * Returns a streaming {@code Iterator<Map<String, Object>>}. Memory-efficient lazy loading for large result sets.
+ * Uses pageSize for internal fetch size. Pagination handled automatically. Does NOT set NextPageToken or JobId
+ * headers.
+ * <p>
+ * Use when: processing large result sets, one-time row processing, streaming to another endpoint.
+ */
+ STREAM_LIST
+}
diff --git a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/integration/sql/SqlQuerySelectAndStreamIT.java b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/integration/sql/SqlQuerySelectAndStreamIT.java
new file mode 100644
index 000000000000..fc22c0bbbf8e
--- /dev/null
+++ b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/integration/sql/SqlQuerySelectAndStreamIT.java
@@ -0,0 +1,142 @@
+/*
+ * 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.google.bigquery.integration.sql;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Message;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.google.bigquery.GoogleBigQueryConstants;
+import org.apache.camel.component.google.bigquery.integration.BigQueryITSupport;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.condition.EnabledIf;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.InstanceOfAssertFactories.LIST;
+import static org.assertj.core.api.InstanceOfAssertFactories.MAP;
+import static org.assertj.core.api.InstanceOfAssertFactories.iterator;
+import static org.assertj.core.api.InstanceOfAssertFactories.list;
+import static org.assertj.core.api.InstanceOfAssertFactories.map;
+
+@EnabledIf(value = "org.apache.camel.component.google.bigquery.integration.BigQueryITSupport#hasCredentials",
+ disabledReason = "Credentials were not provided")
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class SqlQuerySelectAndStreamIT extends BigQueryITSupport {
+
+ @EndpointInject("mock:selectResult")
+ private MockEndpoint selectResult;
+
+ @EndpointInject("mock:streamResult")
+ private MockEndpoint streamResult;
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ String baseUri = "google-bigquery-sql:{{project.id}}:classpath:sql/select.sql";
+
+ from("direct:selectListPaged")
+ .toD(baseUri + "?pageSize=1")
+ .to("mock:selectResult");
+
+ from("direct:streamList")
+ .toD(baseUri + "?outputType=STREAM_LIST&pageSize=1")
+ .to("mock:streamResult");
+ }
+ };
+ }
+
+ @Test
+ public void testStreamList() throws Exception {
+ streamResult.reset();
+ streamResult.expectedMessageCount(1);
+ template.sendBody("direct:streamList", null);
+ streamResult.assertIsSatisfied();
+
+ Message streamListMessage = streamResult.getExchanges().get(0).getMessage();
+ Object body = streamListMessage.getBody();
+
+ assertThat(body).asInstanceOf(iterator(Map.class)).satisfies(iterator -> {
+ List<Map<String, Object>> rows = new ArrayList<>();
+ iterator.forEachRemaining(rows::add);
+
+ assertThat(rows).hasSize(2);
+
+ for (int i = 0; i < 2; i++) {
+ assertRowContent(rows.get(i), i + 1);
+ }
+ });
+ }
+
+ @Test
+ public void testSelectListPagination() throws Exception {
+ selectResult.reset();
+ selectResult.expectedMessageCount(1);
+ template.sendBody("direct:selectListPaged", null);
+ selectResult.assertIsSatisfied();
+
+ Message firstPageMessage = selectResult.getExchanges().get(0).getMessage();
+ Object firstPageRows = firstPageMessage.getBody();
+ assertThat(firstPageRows).asInstanceOf(list(Map.class))
+ .hasSize(1)
+ .first(map(String.class, Object.class))
+ .satisfies(row -> assertRowContent(row, 1));
+
+ // Get pagination info
+ String nextPageToken = firstPageMessage.getHeader(GoogleBigQueryConstants.NEXT_PAGE_TOKEN, String.class);
+ Object jobId = firstPageMessage.getHeader(GoogleBigQueryConstants.JOB_ID);
+
+ assertThat(nextPageToken).isNotNull();
+ assertThat(jobId).isNotNull();
+
+ // Query second page using pageToken
+ selectResult.reset();
+ selectResult.expectedMessageCount(1);
+
+ Map<String, Object> headers = new HashMap<>();
+ headers.put(GoogleBigQueryConstants.PAGE_TOKEN, nextPageToken);
+ headers.put(GoogleBigQueryConstants.JOB_ID, jobId);
+ template.sendBodyAndHeaders("direct:selectListPaged", null, headers);
+ selectResult.assertIsSatisfied();
+
+ Message secondPageMessage = selectResult.getExchanges().get(0).getMessage();
+ Object secondPageRows = secondPageMessage.getBody(List.class);
+ assertThat(secondPageRows).asInstanceOf(list(Map.class))
+ .hasSize(1)
+ .first(map(String.class, Object.class))
+ .satisfies(row -> assertRowContent(row, 2));
+ }
+
+ private void assertRowContent(Map<String, Object> row, int seq) {
+ assertThat(row.get("address")).asInstanceOf(MAP)
+ .containsEntry("city", "City" + seq);
+
+ assertThat(row.get("tags")).asInstanceOf(LIST)
+ .contains("tag" + seq);
+
+ assertThat(row.get("contacts")).asInstanceOf(LIST)
+ .hasSize(2)
+ .first(MAP)
+ .containsEntry("value", "user" + seq + "ATexample.com");
+ }
+}
diff --git a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/FieldValueListMapperTest.java b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/FieldValueListMapperTest.java
new file mode 100644
index 000000000000..10fd3b3a8af0
--- /dev/null
+++ b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/FieldValueListMapperTest.java
@@ -0,0 +1,168 @@
+/*
+ * 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.google.bigquery.unit.sql;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import com.google.cloud.bigquery.Field;
+import com.google.cloud.bigquery.FieldList;
+import com.google.cloud.bigquery.FieldValue;
+import com.google.cloud.bigquery.FieldValueList;
+import com.google.cloud.bigquery.StandardSQLTypeName;
+import org.apache.camel.component.google.bigquery.sql.FieldValueListMapper;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class FieldValueListMapperTest {
+
+ @Test
+ public void mapSimpleRow() {
+ Field idField = Field.of("id", StandardSQLTypeName.INT64);
+ Field nameField = Field.of("name", StandardSQLTypeName.STRING);
+ FieldList fieldList = FieldList.of(idField, nameField);
+
+ FieldValueList row = FieldValueList.of(
+ Arrays.asList(
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 123L),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "John Doe")),
+ fieldList);
+
+ FieldValueListMapper mapper = new FieldValueListMapper(fieldList);
+ Map<String, Object> result = mapper.map(row);
+
+ assertNotNull(result);
+ assertEquals(2, result.size());
+ assertEquals(123L, result.get("id"));
+ assertEquals("John Doe", result.get("name"));
+ }
+
+ @Test
+ public void mapRowWithNullValue() {
+ Field idField = Field.of("id", StandardSQLTypeName.INT64);
+ Field nameField = Field.of("name", StandardSQLTypeName.STRING);
+ Field ageField = Field.of("age", StandardSQLTypeName.INT64);
+ FieldList fieldList = FieldList.of(idField, nameField, ageField);
+
+ FieldValueList row = FieldValueList.of(
+ Arrays.asList(
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 123L),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "John Doe"),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, null)),
+ fieldList);
+
+ FieldValueListMapper mapper = new FieldValueListMapper(fieldList);
+ Map<String, Object> result = mapper.map(row);
+
+ assertNotNull(result);
+ assertEquals(3, result.size());
+ assertTrue(result.containsKey("age"), "Should contain age key even if value is null");
+ assertNull(result.get("age"), "Age value should be null");
+ assertEquals(123L, result.get("id"));
+ assertEquals("John Doe", result.get("name"));
+ }
+
+ @Test
+ public void mapRowWithAllNullValues() {
+ Field col1Field = Field.of("col1", StandardSQLTypeName.STRING);
+ Field col2Field = Field.of("col2", StandardSQLTypeName.STRING);
+ FieldList fieldList = FieldList.of(col1Field, col2Field);
+
+ FieldValueList row = FieldValueList.of(
+ Arrays.asList(
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, null),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, null)),
+ fieldList);
+
+ FieldValueListMapper mapper = new FieldValueListMapper(fieldList);
+ Map<String, Object> result = mapper.map(row);
+
+ assertNotNull(result);
+ assertEquals(2, result.size());
+ assertNull(result.get("col1"));
+ assertNull(result.get("col2"));
+ }
+
+ @Test
+ public void mapRowWithMultipleDataTypes() {
+ Field intField = Field.of("int_col", StandardSQLTypeName.INT64);
+ Field stringField = Field.of("string_col", StandardSQLTypeName.STRING);
+ Field floatField = Field.of("float_col", StandardSQLTypeName.FLOAT64);
+ Field boolField = Field.of("bool_col", StandardSQLTypeName.BOOL);
+ FieldList fieldList = FieldList.of(intField, stringField, floatField, boolField);
+
+ FieldValueList row = FieldValueList.of(
+ Arrays.asList(
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 42L),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "test"),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 3.14),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, true)),
+ fieldList);
+
+ FieldValueListMapper mapper = new FieldValueListMapper(fieldList);
+ Map<String, Object> result = mapper.map(row);
+
+ assertNotNull(result);
+ assertEquals(4, result.size());
+ assertEquals(42L, result.get("int_col"));
+ assertEquals("test", result.get("string_col"));
+ assertEquals(3.14, result.get("float_col"));
+ assertEquals(true, result.get("bool_col"));
+ }
+
+ @Test
+ public void mapEmptyRow() {
+ FieldList fieldList = FieldList.of();
+
+ FieldValueList row = FieldValueList.of(List.of(), fieldList);
+
+ FieldValueListMapper mapper = new FieldValueListMapper(fieldList);
+ Map<String, Object> result = mapper.map(row);
+
+ assertNotNull(result);
+ assertTrue(result.isEmpty(), "Result should be empty map");
+ }
+
+ @Test
+ public void mapRowWithSpecialCharactersInFieldNames() {
+ Field field1 = Field.of("field_with_underscore", StandardSQLTypeName.STRING);
+ Field field2 = Field.of("field-with-dash", StandardSQLTypeName.STRING);
+ Field field3 = Field.of("fieldWithCamelCase", StandardSQLTypeName.STRING);
+ FieldList fieldList = FieldList.of(field1, field2, field3);
+
+ FieldValueList row = FieldValueList.of(
+ Arrays.asList(
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "value1"),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "value2"),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "value3")),
+ fieldList);
+
+ FieldValueListMapper mapper = new FieldValueListMapper(fieldList);
+ Map<String, Object> result = mapper.map(row);
+
+ assertNotNull(result);
+ assertEquals(3, result.size());
+ assertEquals("value1", result.get("field_with_underscore"));
+ assertEquals("value2", result.get("field-with-dash"));
+ assertEquals("value3", result.get("fieldWithCamelCase"));
+ }
+}
diff --git a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerBaseTest.java b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerBaseTest.java
index 0f394ace2efd..0a84b28acef2 100644
--- a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerBaseTest.java
+++ b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerBaseTest.java
@@ -42,6 +42,7 @@ public abstract class GoogleBigQuerySQLProducerBaseTest extends CamelTestSupport
protected TableResult tableResult;
protected Job job;
protected JobStatistics.QueryStatistics statistics;
+ protected QueryJobConfiguration jobConfiguration;
protected GoogleBigQuerySQLProducer createAndStartProducer() {
configuration.setProjectId(projectId);
@@ -57,11 +58,15 @@ public abstract class GoogleBigQuerySQLProducerBaseTest extends CamelTestSupport
tableResult = mock(TableResult.class);
job = mock(Job.class);
statistics = mock(JobStatistics.QueryStatistics.class);
+ jobConfiguration = mock(QueryJobConfiguration.class);
when(bigquery.query(any(QueryJobConfiguration.class), any(JobId.class))).thenReturn(tableResult);
when(bigquery.create(any(JobInfo.class))).thenReturn(job);
+ when(bigquery.getJob(any(JobId.class))).thenReturn(job);
when(job.waitFor()).thenReturn(job);
when(job.getQueryResults()).thenReturn(tableResult);
when(job.getStatistics()).thenReturn(statistics);
+ when(job.<QueryJobConfiguration> getConfiguration()).thenReturn(jobConfiguration);
when(statistics.getNumDmlAffectedRows()).thenReturn(1L);
+ when(jobConfiguration.getQuery()).thenReturn(sql);
}
}
diff --git a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerPaginationTest.java b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerPaginationTest.java
new file mode 100644
index 000000000000..1ad1b565a0d8
--- /dev/null
+++ b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerPaginationTest.java
@@ -0,0 +1,177 @@
+/*
+ * 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.google.bigquery.unit.sql;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+
+import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.Field;
+import com.google.cloud.bigquery.FieldList;
+import com.google.cloud.bigquery.FieldValue;
+import com.google.cloud.bigquery.FieldValueList;
+import com.google.cloud.bigquery.JobId;
+import com.google.cloud.bigquery.JobStatistics;
+import com.google.cloud.bigquery.Schema;
+import com.google.cloud.bigquery.StandardSQLTypeName;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.component.google.bigquery.GoogleBigQueryConstants;
+import org.apache.camel.component.google.bigquery.sql.OutputType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class GoogleBigQuerySQLProducerPaginationTest extends GoogleBigQuerySQLProducerBaseTest {
+
+ @BeforeEach
+ public void init() throws Exception {
+ sql = "SELECT id, name FROM testDataset.testTable LIMIT 100";
+ setupBigqueryMock();
+ when(statistics.getStatementType()).thenReturn(JobStatistics.QueryStatistics.StatementType.SELECT);
+
+ when(job.getQueryResults(any(BigQuery.QueryResultsOption[].class))).thenReturn(tableResult);
+
+ producer = createAndStartProducer();
+ }
+
+ @Test
+ public void fetchDataWithPageSizeInConfiguration() throws Exception {
+ long expectedPageSize = 10;
+ configuration.setPageSize(expectedPageSize);
+
+ Exchange exchange = createExchangeWithBody(null);
+ producer.process(exchange);
+
+ ArgumentCaptor<BigQuery.QueryResultsOption[]> dataCaptor = ArgumentCaptor.forClass(BigQuery.QueryResultsOption[].class);
+ verify(job).getQueryResults(dataCaptor.capture());
+ BigQuery.QueryResultsOption[] capturedOptions = dataCaptor.getValue();
+
+ assertNotNull(capturedOptions, "Options should have been captured");
+ assertEquals(1, capturedOptions.length, "Should have one option for page size");
+ assertEquals(BigQuery.QueryResultsOption.pageSize(expectedPageSize), capturedOptions[0]);
+ }
+
+ @Test
+ public void fetchDataWithPageSizeZeroDoesNotSetOption() throws Exception {
+ configuration.setPageSize(0);
+
+ Exchange exchange = createExchangeWithBody(null);
+ producer.process(exchange);
+
+ ArgumentCaptor<BigQuery.QueryResultsOption[]> dataCaptor = ArgumentCaptor.forClass(BigQuery.QueryResultsOption[].class);
+ verify(job).getQueryResults(dataCaptor.capture());
+ BigQuery.QueryResultsOption[] capturedOptions = dataCaptor.getValue();
+
+ assertNotNull(capturedOptions, "Options should have been captured");
+ assertEquals(0, capturedOptions.length, "Should have no options when page size is 0");
+ }
+
+ @Test
+ public void selectListSetsNextPageTokenAndJobId() throws Exception {
+ configuration.setOutputType(OutputType.SELECT_LIST);
+
+ Exchange exchange = createExchangeWithBody(null);
+ producer.process(exchange);
+
+ Message message = exchange.getMessage();
+
+ assertEquals("next_page_token_123", message.getHeader(GoogleBigQueryConstants.NEXT_PAGE_TOKEN));
+ assertEquals(JobId.of("JOB_ID"), message.getHeader(GoogleBigQueryConstants.JOB_ID));
+
+ List<Map<String, Object>> rows = message.getBody(List.class);
+ assertNotNull(rows);
+ assertEquals(2, rows.size());
+ }
+
+ @Test
+ public void streamListDoesNotSetPaginationHeaders() throws Exception {
+ configuration.setOutputType(OutputType.STREAM_LIST);
+
+ Exchange exchange = createExchangeWithBody(null);
+ producer.process(exchange);
+
+ Message message = exchange.getMessage();
+
+ assertNull(message.getHeader(GoogleBigQueryConstants.NEXT_PAGE_TOKEN));
+ assertNull(message.getHeader(GoogleBigQueryConstants.JOB_ID));
+ }
+
+ @Test
+ public void combinedPageSizeAndPageToken() throws Exception {
+ long expectedPageSize = 50;
+ String expectedToken = "start_token";
+ configuration.setPageSize(expectedPageSize);
+ configuration.setPageToken(expectedToken);
+
+ Exchange exchange = createExchangeWithBody(null);
+ producer.process(exchange);
+
+ ArgumentCaptor<BigQuery.QueryResultsOption[]> dataCaptor = ArgumentCaptor.forClass(BigQuery.QueryResultsOption[].class);
+ verify(job).getQueryResults(dataCaptor.capture());
+ BigQuery.QueryResultsOption[] capturedOptions = dataCaptor.getValue();
+
+ assertNotNull(capturedOptions, "Options should have been captured");
+ assertEquals(2, capturedOptions.length, "Should have two options for page token and page size");
+
+ boolean hasPageToken = Arrays.stream(capturedOptions)
+ .anyMatch(opt -> opt.equals(BigQuery.QueryResultsOption.pageToken(expectedToken)));
+ boolean hasPageSize = Arrays.stream(capturedOptions)
+ .anyMatch(opt -> opt.equals(BigQuery.QueryResultsOption.pageSize(expectedPageSize)));
+
+ assertTrue(hasPageToken, "Should contain page token option");
+ assertTrue(hasPageSize, "Should contain page size option");
+ }
+
+ @Override
+ protected void setupBigqueryMock() throws Exception {
+ super.setupBigqueryMock();
+
+ Field idField = Field.of("id", StandardSQLTypeName.INT64);
+ Field nameField = Field.of("name", StandardSQLTypeName.STRING);
+ Schema schema = Schema.of(idField, nameField);
+
+ FieldList fieldList = FieldList.of(idField, nameField);
+ FieldValueList row1 = FieldValueList.of(
+ Arrays.asList(
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 1L),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "Row1")),
+ fieldList);
+ FieldValueList row2 = FieldValueList.of(
+ Arrays.asList(
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 2L),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "Row2")),
+ fieldList);
+
+ List<FieldValueList> rows = Arrays.asList(row1, row2);
+
+ when(tableResult.getSchema()).thenReturn(schema);
+ when(tableResult.getValues()).thenReturn(rows);
+ when(tableResult.iterateAll()).thenReturn(rows::iterator);
+ when(tableResult.getNextPageToken()).thenReturn("next_page_token_123");
+ when(job.getJobId()).thenReturn(JobId.of("JOB_ID"));
+ }
+}
diff --git a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerSelectTest.java b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerSelectTest.java
index 2e7043eb5e70..174f0515ebaf 100644
--- a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerSelectTest.java
+++ b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerSelectTest.java
@@ -20,16 +20,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
-import com.google.cloud.bigquery.BigQuery;
-import com.google.cloud.bigquery.Field;
-import com.google.cloud.bigquery.FieldList;
-import com.google.cloud.bigquery.FieldValue;
-import com.google.cloud.bigquery.FieldValueList;
-import com.google.cloud.bigquery.JobInfo;
-import com.google.cloud.bigquery.QueryJobConfiguration;
-import com.google.cloud.bigquery.Schema;
-import com.google.cloud.bigquery.StandardSQLTypeName;
-import com.google.cloud.bigquery.TableResult;
+import com.google.cloud.bigquery.*;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.component.google.bigquery.GoogleBigQueryConstants;
@@ -79,6 +70,7 @@ public class GoogleBigQuerySQLProducerSelectTest extends GoogleBigQuerySQLProduc
assertEquals("Bob", secondRow.get("name"));
assertEquals("PAGE_TWO_TOKEN", message.getHeader(GoogleBigQueryConstants.NEXT_PAGE_TOKEN));
+ assertEquals(JobId.of("JOB_ID"), message.getHeader(GoogleBigQueryConstants.JOB_ID));
}
@Test
@@ -107,6 +99,7 @@ public class GoogleBigQuerySQLProducerSelectTest extends GoogleBigQuerySQLProduc
assertEquals("Delta", secondRow.get("name"));
assertEquals("PAGE_THREE_TOKEN", message.getHeader(GoogleBigQueryConstants.NEXT_PAGE_TOKEN));
+ assertEquals(JobId.of("JOB_ID"), message.getHeader(GoogleBigQueryConstants.JOB_ID));
}
@Test
@@ -136,6 +129,7 @@ public class GoogleBigQuerySQLProducerSelectTest extends GoogleBigQuerySQLProduc
assertEquals("Delta", secondRow.get("name"));
assertEquals("PAGE_THREE_TOKEN", message.getHeader(GoogleBigQueryConstants.NEXT_PAGE_TOKEN));
+ assertEquals(JobId.of("JOB_ID"), message.getHeader(GoogleBigQueryConstants.JOB_ID));
}
@Override
@@ -169,11 +163,12 @@ public class GoogleBigQuerySQLProducerSelectTest extends GoogleBigQuerySQLProduc
when(tableResult.getSchema()).thenReturn(schema);
when(tableResult.getValues()).thenReturn(Arrays.asList(row1, row2));
when(tableResult.getNextPageToken()).thenReturn("PAGE_TWO_TOKEN");
- when(statistics.getNumDmlAffectedRows()).thenReturn(null);
+ when(statistics.getStatementType()).thenReturn(JobStatistics.QueryStatistics.StatementType.SELECT);
// Mock paging
nextPageTableResult = mock(TableResult.class);
when(job.getQueryResults(any(BigQuery.QueryResultsOption.class))).thenReturn(nextPageTableResult);
+ when(job.getJobId()).thenReturn(JobId.of("JOB_ID"));
when(nextPageTableResult.getSchema()).thenReturn(schema);
when(nextPageTableResult.getValues()).thenReturn(Arrays.asList(row3, row4));
when(nextPageTableResult.getNextPageToken()).thenReturn("PAGE_THREE_TOKEN");
diff --git a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerStreamListTest.java b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerStreamListTest.java
new file mode 100644
index 000000000000..ffbbc8f20ad5
--- /dev/null
+++ b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerStreamListTest.java
@@ -0,0 +1,193 @@
+/*
+ * 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.google.bigquery.unit.sql;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import com.google.cloud.bigquery.Field;
+import com.google.cloud.bigquery.FieldList;
+import com.google.cloud.bigquery.FieldValue;
+import com.google.cloud.bigquery.FieldValueList;
+import com.google.cloud.bigquery.JobStatistics;
+import com.google.cloud.bigquery.Schema;
+import com.google.cloud.bigquery.StandardSQLTypeName;
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.component.google.bigquery.sql.OutputType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+
+public class GoogleBigQuerySQLProducerStreamListTest extends GoogleBigQuerySQLProducerBaseTest {
+
+ @BeforeEach
+ public void init() throws Exception {
+ sql = "SELECT id, name, age FROM testDataset.testTable";
+ setupBigqueryMock();
+ when(statistics.getStatementType()).thenReturn(JobStatistics.QueryStatistics.StatementType.SELECT);
+ configuration.setOutputType(OutputType.STREAM_LIST);
+ producer = createAndStartProducer();
+ }
+
+ @Test
+ public void fetchDataFromBigQueryAsStreamList() throws Exception {
+ Exchange exchange = createExchangeWithBody(null);
+ producer.process(exchange);
+
+ Message message = exchange.getMessage();
+ Object body = message.getBody();
+ assertNotNull(body);
+ assertInstanceOf(Iterator.class, body);
+
+ Iterator<Map<String, Object>> iterator = (Iterator<Map<String, Object>>) body;
+ assertTrue(iterator.hasNext(), "Iterator should have data");
+
+ List<Map<String, Object>> rows = new ArrayList<>();
+ iterator.forEachRemaining(rows::add);
+
+ assertEquals(3, rows.size());
+
+ Map<String, Object> firstRow = rows.get(0);
+ assertEquals(1L, firstRow.get("id"));
+ assertEquals("Alice", firstRow.get("name"));
+ assertEquals(25L, firstRow.get("age"));
+
+ Map<String, Object> secondRow = rows.get(1);
+ assertEquals(2L, secondRow.get("id"));
+ assertEquals("Bob", secondRow.get("name"));
+ assertEquals(30L, secondRow.get("age"));
+
+ Map<String, Object> thirdRow = rows.get(2);
+ assertEquals(3L, thirdRow.get("id"));
+ assertEquals("Charlie", thirdRow.get("name"));
+ assertEquals(35L, thirdRow.get("age"));
+ }
+
+ @Test
+ public void streamListWithNullValues() throws Exception {
+ Exchange exchange = createExchangeWithBody(null);
+ producer.process(exchange);
+
+ Message message = exchange.getMessage();
+ Iterator<Map<String, Object>> iterator = (Iterator<Map<String, Object>>) message.getBody();
+ assertNotNull(iterator);
+
+ List<Map<String, Object>> rows = new ArrayList<>();
+ iterator.forEachRemaining(row -> rows.add(row));
+
+ assertEquals(3, rows.size());
+
+ Map<String, Object> thirdRow = rows.get(2);
+ assertTrue(thirdRow.containsKey("age"), "Row should contain age key even if null");
+ }
+
+ @Test
+ public void streamListWithEmptyResult() throws Exception {
+ when(tableResult.getValues()).thenReturn(List.of());
+ when(tableResult.iterateAll()).thenReturn(Collections::emptyIterator);
+
+ Exchange exchange = createExchangeWithBody(null);
+ producer.process(exchange);
+
+ Message message = exchange.getMessage();
+ Iterator<?> iterator = (Iterator<?>) message.getBody();
+ assertNotNull(iterator);
+ assertFalse(iterator.hasNext(), "Iterator should be empty");
+ }
+
+ @Test
+ public void streamListDoesNotSetNextPageTokenHeader() throws Exception {
+ Exchange exchange = createExchangeWithBody(null);
+ producer.process(exchange);
+
+ Message message = exchange.getMessage();
+ assertFalse(message.getHeaders().containsKey("CamelGoogleBigQueryNextPageToken"),
+ "STREAM_LIST mode should not set NextPageToken header");
+ assertFalse(message.getHeaders().containsKey("CamelGoogleBigQueryJobId"),
+ "STREAM_LIST mode should not set JobId header");
+ }
+
+ @Test
+ public void streamListHandlesMultipleDataTypes() throws Exception {
+ Exchange exchange = createExchangeWithBody(null);
+ producer.process(exchange);
+
+ Message message = exchange.getMessage();
+ Iterator<Map<String, Object>> iterator = (Iterator<Map<String, Object>>) message.getBody();
+
+ List<Map<String, Object>> rows = new ArrayList<>();
+ iterator.forEachRemaining(row -> rows.add(row));
+
+ assertEquals(3, rows.size());
+
+ Map<String, Object> firstRow = rows.get(0);
+ assertInstanceOf(Long.class, firstRow.get("id"), "ID should be Long");
+ assertInstanceOf(String.class, firstRow.get("name"), "Name should be String");
+ assertInstanceOf(Long.class, firstRow.get("age"), "Age should be Long");
+ }
+
+ @Override
+ protected void setupBigqueryMock() throws Exception {
+ super.setupBigqueryMock();
+
+ Field idField = Field.of("id", StandardSQLTypeName.INT64);
+ Field nameField = Field.of("name", StandardSQLTypeName.STRING);
+ Field ageField = Field.of("age", StandardSQLTypeName.INT64);
+ Schema schema = Schema.of(idField, nameField, ageField);
+
+ FieldList fieldList = FieldList.of(idField, nameField, ageField);
+
+ FieldValueList row1 = FieldValueList.of(
+ Arrays.asList(
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 1L),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "Alice"),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 25L)),
+ fieldList);
+
+ FieldValueList row2 = FieldValueList.of(
+ Arrays.asList(
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 2L),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "Bob"),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 30L)),
+ fieldList);
+
+ FieldValueList row3 = FieldValueList.of(
+ Arrays.asList(
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 3L),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, "Charlie"),
+ FieldValue.of(FieldValue.Attribute.PRIMITIVE, 35L)),
+ fieldList);
+
+ List<FieldValueList> rows = Arrays.asList(row1, row2, row3);
+
+ when(tableResult.getSchema()).thenReturn(schema);
+ when(tableResult.getValues()).thenReturn(rows);
+ when(tableResult.iterateAll()).thenReturn(rows::iterator);
+ when(statistics.getNumDmlAffectedRows()).thenReturn(null);
+ }
+}
diff --git a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerWithParamersTest.java b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerWithParamersTest.java
index c9ee2732a048..2ed1daba068b 100644
--- a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerWithParamersTest.java
+++ b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerWithParamersTest.java
@@ -20,11 +20,7 @@ import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;
-import com.google.cloud.bigquery.JobId;
-import com.google.cloud.bigquery.JobInfo;
-import com.google.cloud.bigquery.QueryJobConfiguration;
-import com.google.cloud.bigquery.QueryParameterValue;
-import com.google.cloud.bigquery.StandardSQLTypeName;
+import com.google.cloud.bigquery.*;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.RuntimeExchangeException;
@@ -38,6 +34,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class GoogleBigQuerySQLProducerWithParamersTest extends GoogleBigQuerySQLProducerBaseTest {
@@ -45,6 +42,7 @@ public class GoogleBigQuerySQLProducerWithParamersTest extends GoogleBigQuerySQL
public void init() throws Exception {
sql = "insert into testDatasetId.testTableId(id, data) values(@id, @data)";
setupBigqueryMock();
+ when(statistics.getStatementType()).thenReturn(JobStatistics.QueryStatistics.StatementType.INSERT);
producer = createAndStartProducer();
}
@@ -174,20 +172,11 @@ public class GoogleBigQuerySQLProducerWithParamersTest extends GoogleBigQuerySQL
producer.process(exchange);
- ArgumentCaptor<JobInfo> dataCaptor = ArgumentCaptor.forClass(JobInfo.class);
- verify(bigquery).create(dataCaptor.capture());
+ ArgumentCaptor<JobId> dataCaptor = ArgumentCaptor.forClass(JobId.class);
+ verify(bigquery).getJob(dataCaptor.capture());
- QueryJobConfiguration request = dataCaptor.getValue().getConfiguration();
- assertEquals(sql, request.getQuery());
-
- Map<String, QueryParameterValue> namedParameters = request.getNamedParameters();
- assertEquals(2, namedParameters.size());
-
- assertTrue(namedParameters.containsKey("id"));
- assertEquals("100", namedParameters.get("id").getValue(), "Body data must have higher priority");
-
- assertTrue(namedParameters.containsKey("data"));
- assertEquals("some data", namedParameters.get("data").getValue());
+ JobId request = dataCaptor.getValue();
+ assertEquals(jobId, request);
}
@Test
diff --git a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerWithPatternTest.java b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerWithPatternTest.java
index 2e838d057052..45bc6bf2ba30 100644
--- a/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerWithPatternTest.java
+++ b/components/camel-google/camel-google-bigquery/src/test/java/org/apache/camel/component/google/bigquery/unit/sql/GoogleBigQuerySQLProducerWithPatternTest.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.google.bigquery.unit.sql;
import java.util.List;
import com.google.cloud.bigquery.JobInfo;
+import com.google.cloud.bigquery.JobStatistics;
import com.google.cloud.bigquery.QueryJobConfiguration;
import org.apache.camel.Exchange;
import org.junit.jupiter.api.BeforeEach;
@@ -27,6 +28,7 @@ import org.mockito.ArgumentCaptor;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class GoogleBigQuerySQLProducerWithPatternTest extends GoogleBigQuerySQLProducerBaseTest {
@@ -34,6 +36,7 @@ public class GoogleBigQuerySQLProducerWithPatternTest extends GoogleBigQuerySQLP
public void init() throws Exception {
sql = "insert into ${testDatasetId}.testTableId(id, data) values(1, 'test')";
setupBigqueryMock();
+ when(statistics.getStatementType()).thenReturn(JobStatistics.QueryStatistics.StatementType.INSERT);
producer = createAndStartProducer();
}
diff --git a/components/camel-google/camel-google-bigquery/src/test/resources/sql/select.sql b/components/camel-google/camel-google-bigquery/src/test/resources/sql/select.sql
new file mode 100644
index 000000000000..b03bfbf040a4
--- /dev/null
+++ b/components/camel-google/camel-google-bigquery/src/test/resources/sql/select.sql
@@ -0,0 +1,33 @@
+--
+-- 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.
+--
+
+SELECT * FROM UNNEST([
+ STRUCT(
+ 1 AS seq,
+ 'Row 1' AS name,
+ STRUCT('1 Main St' AS street, 'City1' AS city, '90001' AS zip) AS address,
+ ['tag1', 'test'] AS tags,
+ [STRUCT('email' AS type, 'user1ATexample.com' AS value), STRUCT('phone' AS type, '+1-555-0001' AS value)] AS contacts
+ ),
+ STRUCT(
+ 2 AS seq,
+ 'Row 2' AS name,
+ STRUCT('2 Main St' AS street, 'City2' AS city, '90002' AS zip) AS address,
+ ['tag2', 'test'] AS tags,
+ [STRUCT('email' AS type, 'user2ATexample.com' AS value), STRUCT('phone' AS type, '+1-555-0002' AS value)] AS contacts
+ )
+]) ORDER BY seq
diff --git a/components/camel-jdbc/src/main/java/org/apache/camel/component/jdbc/StreamListIterator.java b/components/camel-jdbc/src/main/java/org/apache/camel/component/jdbc/JdbcBeanMapper.java
similarity index 63%
rename from components/camel-jdbc/src/main/java/org/apache/camel/component/jdbc/StreamListIterator.java
rename to components/camel-jdbc/src/main/java/org/apache/camel/component/jdbc/JdbcBeanMapper.java
index dc0688ace81d..70b4a8aee023 100644
--- a/components/camel-jdbc/src/main/java/org/apache/camel/component/jdbc/StreamListIterator.java
+++ b/components/camel-jdbc/src/main/java/org/apache/camel/component/jdbc/JdbcBeanMapper.java
@@ -17,57 +17,34 @@
package org.apache.camel.component.jdbc;
import java.sql.SQLException;
-import java.util.Iterator;
import java.util.Map;
-import java.util.function.Consumer;
import org.apache.camel.CamelContext;
+import org.apache.camel.support.RowMapper;
import static org.apache.camel.component.jdbc.JdbcHelper.newBeanInstance;
-public final class StreamListIterator implements Iterator {
-
+public class JdbcBeanMapper implements RowMapper<Map<String, Object>, Object> {
private final CamelContext camelContext;
private final String outputClass;
private final BeanRowMapper beanRowMapper;
- private final Iterator delegate;
- public StreamListIterator(CamelContext camelContext, String outputClass,
- BeanRowMapper beanRowMapper, Iterator delegate) {
+ public JdbcBeanMapper(CamelContext camelContext, String outputClass, BeanRowMapper beanRowMapper) {
this.camelContext = camelContext;
this.outputClass = outputClass;
this.beanRowMapper = beanRowMapper;
- this.delegate = delegate;
}
@Override
- public boolean hasNext() {
- return delegate.hasNext();
- }
-
- @Override
- public Object next() {
- Object answer;
- Map row = (Map) delegate.next();
+ public Object map(Map<String, Object> row) {
if (row != null && outputClass != null) {
try {
- answer = newBeanInstance(camelContext, outputClass, beanRowMapper, row);
+ return newBeanInstance(camelContext, outputClass, beanRowMapper, row);
} catch (SQLException e) {
throw new RuntimeException(e);
}
} else {
- answer = row;
+ return row;
}
- return answer;
- }
-
- @Override
- public void remove() {
- delegate.remove();
- }
-
- @Override
- public void forEachRemaining(Consumer action) {
- delegate.forEachRemaining(action);
}
}
diff --git a/components/camel-jdbc/src/main/java/org/apache/camel/component/jdbc/JdbcProducer.java b/components/camel-jdbc/src/main/java/org/apache/camel/component/jdbc/JdbcProducer.java
index 494c115715dd..d2cf535c525b 100644
--- a/components/camel-jdbc/src/main/java/org/apache/camel/component/jdbc/JdbcProducer.java
+++ b/components/camel-jdbc/src/main/java/org/apache/camel/component/jdbc/JdbcProducer.java
@@ -33,6 +33,7 @@ import javax.sql.DataSource;
import org.apache.camel.Exchange;
import org.apache.camel.support.DefaultProducer;
import org.apache.camel.support.PropertyBindingSupport;
+import org.apache.camel.support.StreamListIterator;
import org.apache.camel.support.SynchronizationAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -392,10 +393,12 @@ public class JdbcProducer extends DefaultProducer {
JdbcOutputType outputType = getEndpoint().getOutputType();
exchange.getMessage().setHeader(JdbcConstants.JDBC_COLUMN_NAMES, iterator.getColumnNames());
if (outputType == JdbcOutputType.StreamList) {
+ var beanMapper = new JdbcBeanMapper(
+ getEndpoint().getCamelContext(),
+ getEndpoint().getOutputClass(),
+ getEndpoint().getBeanRowMapper());
exchange.getMessage()
- .setBody(new StreamListIterator(
- getEndpoint().getCamelContext(), getEndpoint().getOutputClass(), getEndpoint().getBeanRowMapper(),
- iterator));
+ .setBody(new StreamListIterator<>(beanMapper, iterator));
// do not close resources as we are in streaming mode
answer = false;
} else if (outputType == JdbcOutputType.SelectList) {
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/RowMapper.java b/core/camel-support/src/main/java/org/apache/camel/support/RowMapper.java
new file mode 100644
index 000000000000..fc24cd33b61f
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/RowMapper.java
@@ -0,0 +1,37 @@
+/*
+ * 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.support;
+
+/**
+ * Maps rows from one type to another. Commonly used for converting database rows, API responses, or structured data
+ * into domain objects or maps.
+ * <p>
+ * Implementations should be stateless and thread-safe when reused across multiple rows.
+ *
+ * @param <T> the input type (e.g., ResultSet row, FieldValueList)
+ * @param <R> the output type (e.g., Map, domain object)
+ */
+public interface RowMapper<T, R> {
+
+ /**
+ * Maps a value from type T to type R.
+ *
+ * @param value the input value to be mapped
+ * @return the mapped output value
+ */
+ R map(T value);
+}
diff --git a/core/camel-support/src/main/java/org/apache/camel/support/StreamListIterator.java b/core/camel-support/src/main/java/org/apache/camel/support/StreamListIterator.java
new file mode 100644
index 000000000000..fca9ad59f944
--- /dev/null
+++ b/core/camel-support/src/main/java/org/apache/camel/support/StreamListIterator.java
@@ -0,0 +1,53 @@
+/*
+ * 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.support;
+
+import java.util.Iterator;
+
+/**
+ * Iterator that applies a {@link RowMapper} transformation to each element lazily. Useful for memory-efficient
+ * processing of large result sets from databases or APIs.
+ *
+ * @param <T> the input type from the delegate iterator
+ * @param <E> the output type after applying the row mapper
+ */
+public final class StreamListIterator<T, E> implements Iterator<E> {
+
+ private final RowMapper<T, E> rowMapper;
+ private final Iterator<T> delegate;
+
+ public StreamListIterator(RowMapper<T, E> rowMapper, Iterator<T> delegate) {
+ this.rowMapper = rowMapper;
+ this.delegate = delegate;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return delegate.hasNext();
+ }
+
+ @Override
+ public E next() {
+ T row = delegate.next();
+ return rowMapper.map(row);
+ }
+
+ @Override
+ public void remove() {
+ delegate.remove();
+ }
+}
diff --git a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GoogleBigQuerySQLEndpointBuilderFactory.java b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GoogleBigQuerySQLEndpointBuilderFactory.java
index 08c11bccf553..ad6ce86b7e24 100644
--- a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GoogleBigQuerySQLEndpointBuilderFactory.java
+++ b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/GoogleBigQuerySQLEndpointBuilderFactory.java
@@ -76,6 +76,68 @@ public interface GoogleBigQuerySQLEndpointBuilderFactory {
doSetProperty("connectionFactory", connectionFactory);
return this;
}
+ /**
+ * Output mode type for select queries.
+ *
+ * The option is a:
+ * <code>org.apache.camel.component.google.bigquery.sql.OutputType</code> type.
+ *
+ * Default: SELECT_LIST
+ * Group: producer
+ *
+ * @param outputType the value to set
+ * @return the dsl builder
+ */
+ default GoogleBigQuerySQLEndpointBuilder outputType(org.apache.camel.component.google.bigquery.sql.OutputType outputType) {
+ doSetProperty("outputType", outputType);
+ return this;
+ }
+ /**
+ * Output mode type for select queries.
+ *
+ * The option will be converted to a
+ * <code>org.apache.camel.component.google.bigquery.sql.OutputType</code> type.
+ *
+ * Default: SELECT_LIST
+ * Group: producer
+ *
+ * @param outputType the value to set
+ * @return the dsl builder
+ */
+ default GoogleBigQuerySQLEndpointBuilder outputType(String outputType) {
+ doSetProperty("outputType", outputType);
+ return this;
+ }
+ /**
+ * Size of a page to request. Unset by default.
+ *
+ * The option is a: <code>long</code> type.
+ *
+ * Default: 0
+ * Group: producer
+ *
+ * @param pageSize the value to set
+ * @return the dsl builder
+ */
+ default GoogleBigQuerySQLEndpointBuilder pageSize(long pageSize) {
+ doSetProperty("pageSize", pageSize);
+ return this;
+ }
+ /**
+ * Size of a page to request. Unset by default.
+ *
+ * The option will be converted to a <code>long</code> type.
+ *
+ * Default: 0
+ * Group: producer
+ *
+ * @param pageSize the value to set
+ * @return the dsl builder
+ */
+ default GoogleBigQuerySQLEndpointBuilder pageSize(String pageSize) {
+ doSetProperty("pageSize", pageSize);
+ return this;
+ }
/**
* Page token, returned by a previous call, to request the next page of
* results.