From 9c02e11eac359fce7cbcb973e78f341cc6e6eb57 Mon Sep 17 00:00:00 2001
From: Volodymyr <dgray16@users.noreply.github.com>
Date: Tue, 17 Jan 2023 16:19:56 +0200
Subject: [PATCH] [bq] 0.2 introduce BigQuery interactive reader

---
 spring-batch-bigquery/pom.xml                 |   8 +-
 .../BigQueryInteractiveQueryItemReader.java   | 111 +++++++++++++++++
 ...ueryInteractiveQueryItemReaderBuilder.java | 116 ++++++++++++++++++
 .../writer/BigQueryBaseItemWriter.java        |  75 +++++++++--
 .../writer/BigQueryCsvItemWriter.java         |  23 +++-
 .../writer/BigQueryJsonItemWriter.java        |  25 +++-
 .../builder/BigQueryCsvItemWriterBuilder.java |  46 ++++++-
 .../BigQueryJsonItemWriterBuilder.java        |  46 ++++++-
 .../bigquery/writer/package-info.java         |  14 +--
 .../bigquery/common/BigQueryDataLoader.java   |  20 +--
 .../extensions/bigquery/common/PersonDto.java |   2 +-
 .../bigquery/common/TestConstants.java        |  14 ++-
 .../base/BaseBigQueryIntegrationTest.java     |  38 ++++++
 .../bigquery/integration/package-info.java    |  21 +++-
 ...ueryInteractiveQueryCsvItemReaderTest.java |  68 ++++++++++
 ...eryInteractiveQueryJsonItemReaderTest.java |  68 ++++++++++
 ...CsvJsonInteractiveQueryItemReaderTest.java |  54 ++++++++
 .../writer/BigQueryCsvItemWriterTest.java     |   2 +-
 .../writer/BigQueryJsonItemWriterTest.java    |   2 +-
 .../base/BaseBigQueryItemWriterTest.java      |  44 +------
 .../unit/base/AbstractBigQueryTest.java       |  16 +++
 ...nteractiveQueryItemReaderBuilderTests.java |  67 ++++++++++
 .../BigQueryCsvItemWriterBuilderTests.java    |   2 +-
 .../BigQueryJsonItemWriterBuilderTests.java   |  10 +-
 24 files changed, 791 insertions(+), 101 deletions(-)
 create mode 100644 spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/reader/BigQueryInteractiveQueryItemReader.java
 create mode 100644 spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/reader/builder/BigQueryInteractiveQueryItemReaderBuilder.java
 create mode 100644 spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/base/BaseBigQueryIntegrationTest.java
 create mode 100644 spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/BigQueryInteractiveQueryCsvItemReaderTest.java
 create mode 100644 spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/BigQueryInteractiveQueryJsonItemReaderTest.java
 create mode 100644 spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/base/BaseCsvJsonInteractiveQueryItemReaderTest.java
 create mode 100644 spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/reader/builder/BigQueryInteractiveQueryItemReaderBuilderTests.java

diff --git a/spring-batch-bigquery/pom.xml b/spring-batch-bigquery/pom.xml
index 93ae349..9027952 100644
--- a/spring-batch-bigquery/pom.xml
+++ b/spring-batch-bigquery/pom.xml
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-  ~ Copyright 2002-2022 the original author or authors.
+  ~ Copyright 2002-2023 the original author or authors.
   ~
   ~ Licensed under the Apache License, Version 2.0 (the "License");
   ~ you may not use this file except in compliance with the License.
@@ -64,7 +64,7 @@
         <dependency>
             <groupId>com.google.cloud</groupId>
             <artifactId>google-cloud-bigquery</artifactId>
-            <version>2.19.1</version>
+            <version>2.20.2</version>
         </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.dataformat</groupId>
@@ -88,13 +88,13 @@
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter-api</artifactId>
-            <version>5.9.1</version>
+            <version>5.9.2</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
             <artifactId>mockito-core</artifactId>
-            <version>4.9.0</version>
+            <version>5.0.0</version>
             <scope>test</scope>
         </dependency>
         <dependency>
diff --git a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/reader/BigQueryInteractiveQueryItemReader.java b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/reader/BigQueryInteractiveQueryItemReader.java
new file mode 100644
index 0000000..de677fd
--- /dev/null
+++ b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/reader/BigQueryInteractiveQueryItemReader.java
@@ -0,0 +1,111 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.springframework.batch.extensions.bigquery.reader;
+
+import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.FieldValueList;
+import com.google.cloud.bigquery.QueryJobConfiguration;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.util.Assert;
+
+import java.util.Iterator;
+import java.util.Objects;
+
+/**
+ * BigQuery {@link ItemReader} that accepts simple query as the input.
+ * <p>
+ * Internally BigQuery Java library creates a {@link com.google.cloud.bigquery.JobConfiguration.Type#QUERY} job.
+ * Which means that result is coming asynchronously.
+ * <p>
+ * Also, worth mentioning that you should take into account concurrency limits.
+ * <p>
+ * Results of this query by default are stored in a shape of temporary table.
+ *
+ * @param <T> your DTO type
+ * @author Volodymyr Perebykivskyi
+ * @since 0.2.0
+ * @see <a href="https://cloud.google.com/bigquery/docs/running-queries#queries">Interactive queries</a>
+ * @see <a href="https://cloud.google.com/bigquery/quotas#concurrent_rate_interactive_queries">Concurrency limits</a>
+ */
+public class BigQueryInteractiveQueryItemReader<T> implements ItemReader<T>, InitializingBean {
+
+    private final Log logger = LogFactory.getLog(getClass());
+
+    private BigQuery bigQuery;
+    private Converter<FieldValueList, T> rowMapper;
+    private QueryJobConfiguration jobConfiguration;
+    private Iterator<FieldValueList> iterator;
+
+    /**
+     * BigQuery service, responsible for API calls.
+     *
+     * @param bigQuery BigQuery service
+     */
+    public void setBigQuery(BigQuery bigQuery) {
+        this.bigQuery = bigQuery;
+    }
+
+    /**
+     * Row mapper which transforms single BigQuery row into desired type.
+     *
+     * @param rowMapper your row mapper
+     */
+    public void setRowMapper(Converter<FieldValueList, T> rowMapper) {
+        this.rowMapper = rowMapper;
+    }
+
+    /**
+     * Specifies query to run, destination table, etc.
+     *
+     * @param jobConfiguration BigQuery job configuration
+     */
+    public void setJobConfiguration(QueryJobConfiguration jobConfiguration) {
+        this.jobConfiguration = jobConfiguration;
+    }
+
+    @Override
+    public T read() throws Exception {
+        if (Objects.isNull(iterator)) {
+            doOpen();
+        }
+
+        if (logger.isDebugEnabled()) {
+            logger.debug("Reading next element");
+        }
+
+        return iterator.hasNext() ? rowMapper.convert(iterator.next()) : null;
+    }
+
+    private void doOpen() throws Exception {
+        if (logger.isDebugEnabled()) {
+            logger.debug("Executing query");
+        }
+        iterator = bigQuery.query(jobConfiguration).getValues().iterator();
+    }
+
+    @Override
+    public void afterPropertiesSet() {
+        Assert.notNull(this.bigQuery, "BigQuery service must be provided");
+        Assert.notNull(this.rowMapper, "Row mapper must be provided");
+        Assert.notNull(this.jobConfiguration, "Job configuration must be provided");
+    }
+
+}
diff --git a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/reader/builder/BigQueryInteractiveQueryItemReaderBuilder.java b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/reader/builder/BigQueryInteractiveQueryItemReaderBuilder.java
new file mode 100644
index 0000000..556a347
--- /dev/null
+++ b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/reader/builder/BigQueryInteractiveQueryItemReaderBuilder.java
@@ -0,0 +1,116 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.springframework.batch.extensions.bigquery.reader.builder;
+
+import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.FieldValueList;
+import com.google.cloud.bigquery.QueryJobConfiguration;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.batch.extensions.bigquery.reader.BigQueryInteractiveQueryItemReader;
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.util.Assert;
+
+import java.util.Objects;
+
+/**
+ * A builder for {@link BigQueryInteractiveQueryItemReader}.
+ *
+ * @param <T> your DTO type
+ * @author Volodymyr Perebykivskyi
+ * @since 0.2.0
+ * @see <a href="https://github.com/spring-projects/spring-batch-extensions/tree/main/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/reader/builder/BigQueryInteractiveQueryItemReaderBuilderTests.java">Examples</a>
+ */
+public class BigQueryInteractiveQueryItemReaderBuilder<T> {
+
+    private BigQuery bigQuery;
+    private String query;
+    private Converter<FieldValueList, T> rowMapper;
+    private QueryJobConfiguration jobConfiguration;
+
+    /**
+     * BigQuery service, responsible for API calls.
+     *
+     * @param bigQuery BigQuery service
+     * @return {@link BigQueryInteractiveQueryItemReaderBuilder}
+     * @see BigQueryInteractiveQueryItemReader#setBigQuery(BigQuery)
+     */
+    public BigQueryInteractiveQueryItemReaderBuilder<T> bigQuery(BigQuery bigQuery) {
+        this.bigQuery = bigQuery;
+        return this;
+    }
+
+    /**
+     * Schema of the query: {@code SELECT <column> FROM <dataset>.<table>}.
+     * <p>
+     * It is really recommended to use {@code LIMIT n}
+     * because BigQuery charges you for the amount of data that is being processed.
+     *
+     * @param query your query to run
+     * @return {@link BigQueryInteractiveQueryItemReaderBuilder}
+     * @see BigQueryInteractiveQueryItemReader#setJobConfiguration(QueryJobConfiguration)
+     */
+    public BigQueryInteractiveQueryItemReaderBuilder<T> query(String query) {
+        this.query = query;
+        return this;
+    }
+
+    /**
+     * Row mapper which transforms single BigQuery row into desired type.
+     *
+     * @param rowMapper your row mapper
+     * @return {@link BigQueryInteractiveQueryItemReaderBuilder}
+     * @see BigQueryInteractiveQueryItemReader#setRowMapper(Converter)
+     */
+    public BigQueryInteractiveQueryItemReaderBuilder<T> rowMapper(Converter<FieldValueList, T> rowMapper) {
+        this.rowMapper = rowMapper;
+        return this;
+    }
+
+    /**
+     * Specifies query to run, destination table, etc.
+     *
+     * @param jobConfiguration BigQuery job configuration
+     * @return {@link BigQueryInteractiveQueryItemReaderBuilder}
+     * @see BigQueryInteractiveQueryItemReader#setJobConfiguration(QueryJobConfiguration)
+     */
+    public BigQueryInteractiveQueryItemReaderBuilder<T> jobConfiguration(QueryJobConfiguration jobConfiguration) {
+        this.jobConfiguration = jobConfiguration;
+        return this;
+    }
+
+    /**
+     * Please do not forget about {@link BigQueryInteractiveQueryItemReader#afterPropertiesSet()}.
+     *
+     * @return {@link BigQueryInteractiveQueryItemReader}
+     */
+    public BigQueryInteractiveQueryItemReader<T> build() {
+        BigQueryInteractiveQueryItemReader<T> reader = new BigQueryInteractiveQueryItemReader<>();
+
+        reader.setBigQuery(this.bigQuery);
+        reader.setRowMapper(this.rowMapper);
+
+        if (Objects.nonNull(this.jobConfiguration)) {
+            reader.setJobConfiguration(this.jobConfiguration);
+        } else {
+            Assert.isTrue(StringUtils.isNotBlank(this.query), "No query provided");
+            reader.setJobConfiguration(QueryJobConfiguration.newBuilder(this.query).build());
+        }
+
+        return reader;
+    }
+
+}
diff --git a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryBaseItemWriter.java b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryBaseItemWriter.java
index e9578f6..9d721c1 100644
--- a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryBaseItemWriter.java
+++ b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryBaseItemWriter.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -43,11 +43,24 @@ import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
+/**
+ * Base class that holds shared code for JSON and CSV writers.
+ *
+ * @param <T> your DTO type
+ * @author Volodymyr Perebykivskyi
+ * @since 0.1.0
+ */
 public abstract class BigQueryBaseItemWriter<T> implements ItemWriter<T> {
 
+    /** Logger that can be reused */
     protected final Log logger = LogFactory.getLog(getClass());
     private final AtomicLong bigQueryWriteCounter = new AtomicLong();
 
+    /**
+     * Describes what should be written (format) and its destination (table).
+     */
+    protected WriteChannelConfiguration writeChannelConfig;
+
     /**
      * You can specify here some specific dataset configuration, like location.
      * This dataset will be created.
@@ -60,25 +73,50 @@ public abstract class BigQueryBaseItemWriter<T> implements ItemWriter<T> {
      */
     private Consumer<Job> jobConsumer;
 
-    protected WriteChannelConfiguration writeChannelConfig;
     private BigQuery bigQuery;
 
+
+    /**
+     * Fetches table from provided configuration.
+     *
+     * @return {@link Table} that is described in {@link BigQueryBaseItemWriter#writeChannelConfig}
+     */
     protected Table getTable() {
         return this.bigQuery.getTable(this.writeChannelConfig.getDestinationTable());
     }
 
+    /**
+     * Provides additional information about the {@link com.google.cloud.bigquery.Dataset}.
+     *
+     * @param datasetInfo BigQuery dataset info
+     */
     public void setDatasetInfo(DatasetInfo datasetInfo) {
         this.datasetInfo = datasetInfo;
     }
 
+    /**
+     * Callback when {@link Job} will be finished.
+     *
+     * @param consumer your consumer
+     */
     public void setJobConsumer(Consumer<Job> consumer) {
         this.jobConsumer = consumer;
     }
 
+    /**
+     * Describes what should be written (format) and its destination (table).
+     *
+     * @param writeChannelConfig BigQuery channel configuration
+     */
     public void setWriteChannelConfig(WriteChannelConfiguration writeChannelConfig) {
         this.writeChannelConfig = writeChannelConfig;
     }
 
+    /**
+     * BigQuery service, responsible for API calls.
+     *
+     * @param bigQuery BigQuery service
+     */
     public void setBigQuery(BigQuery bigQuery) {
         this.bigQuery = bigQuery;
     }
@@ -153,6 +191,11 @@ public abstract class BigQueryBaseItemWriter<T> implements ItemWriter<T> {
         return this.bigQuery.writer(this.writeChannelConfig);
     }
 
+    /**
+     * Performs common validation for CSV and JSON types.
+     *
+     * @param formatSpecificChecks supplies type-specific validation
+     */
     protected void baseAfterPropertiesSet(Supplier<Void> formatSpecificChecks) {
         Assert.notNull(this.bigQuery, "BigQuery service must be provided");
         Assert.notNull(this.writeChannelConfig, "Write channel configuration must be provided");
@@ -220,14 +263,13 @@ public abstract class BigQueryBaseItemWriter<T> implements ItemWriter<T> {
         return FormatOptions.datastoreBackup().getType().equals(this.writeChannelConfig.getFormat());
     }
 
-    protected boolean isCsv() {
-        return FormatOptions.csv().getType().equals(this.writeChannelConfig.getFormat());
-    }
-
-    protected boolean isJson() {
-        return FormatOptions.json().getType().equals(this.writeChannelConfig.getFormat());
-    }
-
+    /**
+     * Schema can be computed on BigQuery side during upload,
+     * so it is good to know when schema is supplied by user manually.
+     *
+     * @param table BigQuery table
+     * @return {@code true} if BigQuery {@link Table} has schema already described
+     */
     protected boolean tableHasDefinedSchema(Table table) {
         return Optional
                 .ofNullable(table)
@@ -237,7 +279,20 @@ public abstract class BigQueryBaseItemWriter<T> implements ItemWriter<T> {
                 .isPresent();
     }
 
+    /**
+     * Method that setting up metadata about chunk that is being processed. In reality is called once.
+     *
+     * @param items current chunk
+     */
     protected abstract void doInitializeProperties(List<? extends T> items);
+
+    /**
+     * Converts chunk into byte array.
+     * Each data type should be converted with respect to its specification.
+     *
+     * @param items current chunk
+     * @return {@link List<byte[]>} converted list of byte arrays
+     */
     protected abstract List<byte[]> convertObjectsToByteArrays(List<? extends T> items);
 
 }
diff --git a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryCsvItemWriter.java b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryCsvItemWriter.java
index 2143462..bb205cf 100644
--- a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryCsvItemWriter.java
+++ b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryCsvItemWriter.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -32,11 +32,19 @@ import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
+/**
+ * CSV writer for BigQuery.
+ *
+ * @param <T> your DTO type
+ * @author Volodymyr Perebykivskyi
+ * @since 0.2.0
+ * @see <a href="https://en.wikipedia.org/wiki/Comma-separated_values">CSV</a>
+ */
 public class BigQueryCsvItemWriter<T> extends BigQueryBaseItemWriter<T> implements InitializingBean {
 
-    protected Converter<T, byte[]> rowMapper;
-    protected ObjectWriter objectWriter;
-    protected Class itemClass;
+    private Converter<T, byte[]> rowMapper;
+    private ObjectWriter objectWriter;
+    private Class itemClass;
 
     /**
      * Actual type of incoming data can be obtained only in runtime
@@ -55,6 +63,11 @@ public class BigQueryCsvItemWriter<T> extends BigQueryBaseItemWriter<T> implemen
         }
     }
 
+    /**
+     * Row mapper which transforms single BigQuery row into desired type.
+     *
+     * @param rowMapper your row mapper
+     */
     public void setRowMapper(Converter<T, byte[]> rowMapper) {
         this.rowMapper = rowMapper;
     }
@@ -96,7 +109,7 @@ public class BigQueryCsvItemWriter<T> extends BigQueryBaseItemWriter<T> implemen
         });
     }
 
-    protected byte[] mapItemToCsv(T t) {
+    private byte[] mapItemToCsv(T t) {
         byte[] result = null;
         try {
             result = Objects.isNull(rowMapper) ? objectWriter.writeValueAsBytes(t) : rowMapper.convert(t);
diff --git a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryJsonItemWriter.java b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryJsonItemWriter.java
index a53d0d7..9f8e18d 100644
--- a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryJsonItemWriter.java
+++ b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/BigQueryJsonItemWriter.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -32,11 +32,19 @@ import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
 
+/**
+ * JSON writer for BigQuery.
+ *
+ * @param <T> your DTO type
+ * @author Volodymyr Perebykivskyi
+ * @since 0.2.0
+ * @see <a href="https://en.wikipedia.org/wiki/JSON">JSON</a>
+ */
 public class BigQueryJsonItemWriter<T> extends BigQueryBaseItemWriter<T> implements InitializingBean {
 
-    protected Converter<T, byte[]> rowMapper;
-    protected ObjectWriter objectWriter;
-    protected Class itemClass;
+    private Converter<T, byte[]> rowMapper;
+    private ObjectWriter objectWriter;
+    private Class itemClass;
 
     @Override
     protected void doInitializeProperties(List<? extends T> items) {
@@ -52,6 +60,11 @@ public class BigQueryJsonItemWriter<T> extends BigQueryBaseItemWriter<T> impleme
         }
     }
 
+    /**
+     * Converter that transforms a single row into byte array.
+     *
+     * @param rowMapper your JSON row mapper
+     */
     public void setRowMapper(Converter<T, byte[]> rowMapper) {
         this.rowMapper = rowMapper;
     }
@@ -93,7 +106,7 @@ public class BigQueryJsonItemWriter<T> extends BigQueryBaseItemWriter<T> impleme
         });
     }
 
-    protected byte[] mapItemToJson(T t) {
+    private byte[] mapItemToJson(T t) {
         byte[] result = null;
         try {
             result = Objects.isNull(rowMapper) ? objectWriter.writeValueAsBytes(t) : rowMapper.convert(t);
@@ -105,7 +118,7 @@ public class BigQueryJsonItemWriter<T> extends BigQueryBaseItemWriter<T> impleme
     }
 
     /**
-     * BigQuery uses ndjson https://github.com/ndjson/ndjson-spec.
+     * BigQuery uses <a href="https://github.com/ndjson/ndjson-spec">ndjson</a>.
      * It is expected that to pass here JSON line generated by
      * {@link com.fasterxml.jackson.databind.ObjectMapper} or any other JSON parser.
      */
diff --git a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/builder/BigQueryCsvItemWriterBuilder.java b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/builder/BigQueryCsvItemWriterBuilder.java
index 5477afb..81e9c64 100644
--- a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/builder/BigQueryCsvItemWriterBuilder.java
+++ b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/builder/BigQueryCsvItemWriterBuilder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,10 +28,10 @@ import java.util.function.Consumer;
 /**
  * A builder for {@link BigQueryCsvItemWriter}.
  *
+ * @param <T> your DTO type
  * @author Volodymyr Perebykivskyi
  * @since 0.2.0
- * @see BigQueryCsvItemWriter
- * @see <a href="https://github.com/spring-projects/spring-batch-extensions/tree/main/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/builder/BigQueryCsvItemWriterBuilderTests.java">Examples</a>
+ * @see <a href="https://github.com/spring-projects/spring-batch-extensions/tree/main/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryCsvItemWriterBuilderTests.java">Examples</a>
  */
 public class BigQueryCsvItemWriterBuilder<T>  {
 
@@ -42,31 +42,71 @@ public class BigQueryCsvItemWriterBuilder<T>  {
     private WriteChannelConfiguration writeChannelConfig;
     private BigQuery bigQuery;
 
+    /**
+     * Row mapper which transforms single BigQuery row into desired type.
+     *
+     * @param rowMapper your row mapper
+     * @return {@link BigQueryCsvItemWriterBuilder}
+     * @see BigQueryCsvItemWriter#setRowMapper(Converter)
+     */
     public BigQueryCsvItemWriterBuilder<T> rowMapper(Converter<T, byte[]> rowMapper) {
         this.rowMapper = rowMapper;
         return this;
     }
 
+    /**
+     * Provides additional information about the {@link com.google.cloud.bigquery.Dataset}.
+     *
+     * @param datasetInfo BigQuery dataset info
+     * @return {@link BigQueryCsvItemWriterBuilder}
+     * @see BigQueryCsvItemWriter#setDatasetInfo(DatasetInfo)
+     */
     public BigQueryCsvItemWriterBuilder<T> datasetInfo(DatasetInfo datasetInfo) {
         this.datasetInfo = datasetInfo;
         return this;
     }
 
+    /**
+     * Callback when {@link Job} will be finished.
+     *
+     * @param consumer your consumer
+     * @return {@link BigQueryCsvItemWriterBuilder}
+     * @see BigQueryCsvItemWriter#setJobConsumer(Consumer)
+     */
     public BigQueryCsvItemWriterBuilder<T> jobConsumer(Consumer<Job> consumer) {
         this.jobConsumer = consumer;
         return this;
     }
 
+    /**
+     * Describes what should be written (format) and its destination (table).
+     *
+     * @param configuration BigQuery channel configuration
+     * @return {@link BigQueryCsvItemWriterBuilder}
+     * @see BigQueryCsvItemWriter#setWriteChannelConfig(WriteChannelConfiguration)
+     */
     public BigQueryCsvItemWriterBuilder<T> writeChannelConfig(WriteChannelConfiguration configuration) {
         this.writeChannelConfig = configuration;
         return this;
     }
 
+    /**
+     * BigQuery service, responsible for API calls.
+     *
+     * @param bigQuery BigQuery service
+     * @return {@link BigQueryCsvItemWriterBuilder}
+     * @see BigQueryCsvItemWriter#setBigQuery(BigQuery)
+     */
     public BigQueryCsvItemWriterBuilder<T> bigQuery(BigQuery bigQuery) {
         this.bigQuery = bigQuery;
         return this;
     }
 
+    /**
+     * Please do not forget about {@link BigQueryCsvItemWriter#afterPropertiesSet()}.
+     *
+     * @return {@link BigQueryCsvItemWriter}
+     */
     public BigQueryCsvItemWriter<T> build() {
         BigQueryCsvItemWriter<T> writer = new BigQueryCsvItemWriter<>();
 
diff --git a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/builder/BigQueryJsonItemWriterBuilder.java b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/builder/BigQueryJsonItemWriterBuilder.java
index 4a4dbdd..0fdfa51 100644
--- a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/builder/BigQueryJsonItemWriterBuilder.java
+++ b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/builder/BigQueryJsonItemWriterBuilder.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -28,10 +28,10 @@ import java.util.function.Consumer;
 /**
  * A builder for {@link BigQueryJsonItemWriter}.
  *
+ * @param <T> your DTO type
  * @author Volodymyr Perebykivskyi
  * @since 0.2.0
- * @see BigQueryJsonItemWriter
- * @see <a href="https://github.com/spring-projects/spring-batch-extensions/tree/main/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/builder/BigQueryJsonItemWriterBuilderTests.java">Examples</a>
+ * @see <a href="https://github.com/spring-projects/spring-batch-extensions/tree/main/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryJsonItemWriterBuilderTests.java">Examples</a>
  */
 public class BigQueryJsonItemWriterBuilder<T>  {
 
@@ -42,31 +42,71 @@ public class BigQueryJsonItemWriterBuilder<T>  {
     private WriteChannelConfiguration writeChannelConfig;
     private BigQuery bigQuery;
 
+    /**
+     * Converts your DTO into byte array.
+     *
+     * @param rowMapper your mapping
+     * @return {@link BigQueryJsonItemWriter}
+     * @see BigQueryJsonItemWriter#setRowMapper(Converter)
+     */
     public BigQueryJsonItemWriterBuilder<T> rowMapper(Converter<T, byte[]> rowMapper) {
         this.rowMapper = rowMapper;
         return this;
     }
 
+    /**
+     * Provides additional information about the {@link com.google.cloud.bigquery.Dataset}.
+     *
+     * @param datasetInfo BigQuery dataset info
+     * @return {@link BigQueryJsonItemWriter}
+     * @see BigQueryJsonItemWriter#setDatasetInfo(DatasetInfo)
+     */
     public BigQueryJsonItemWriterBuilder<T> datasetInfo(DatasetInfo datasetInfo) {
         this.datasetInfo = datasetInfo;
         return this;
     }
 
+    /**
+     * Callback when {@link Job} will be finished.
+     *
+     * @param consumer your consumer
+     * @return {@link BigQueryJsonItemWriter}
+     * @see BigQueryJsonItemWriter#setJobConsumer(Consumer)
+     */
     public BigQueryJsonItemWriterBuilder<T> jobConsumer(Consumer<Job> consumer) {
         this.jobConsumer = consumer;
         return this;
     }
 
+    /**
+     * Describes what should be written (format) and its destination (table).
+     *
+     * @param configuration BigQuery channel configuration
+     * @return {@link BigQueryJsonItemWriter}
+     * @see BigQueryJsonItemWriter#setWriteChannelConfig(WriteChannelConfiguration)
+     */
     public BigQueryJsonItemWriterBuilder<T> writeChannelConfig(WriteChannelConfiguration configuration) {
         this.writeChannelConfig = configuration;
         return this;
     }
 
+    /**
+     * BigQuery service, responsible for API calls.
+     *
+     * @param bigQuery BigQuery service
+     * @return {@link BigQueryJsonItemWriter}
+     * @see BigQueryJsonItemWriter#setBigQuery(BigQuery)
+     */
     public BigQueryJsonItemWriterBuilder<T> bigQuery(BigQuery bigQuery) {
         this.bigQuery = bigQuery;
         return this;
     }
 
+    /**
+     * Please do not forget about {@link BigQueryJsonItemWriter#afterPropertiesSet()}.
+     *
+     * @return {@link BigQueryJsonItemWriter}
+     */
     public BigQueryJsonItemWriter<T> build() {
         BigQueryJsonItemWriter<T> writer = new BigQueryJsonItemWriter<>();
 
diff --git a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/package-info.java b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/package-info.java
index e085743..e5c61f7 100644
--- a/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/package-info.java
+++ b/spring-batch-bigquery/src/main/java/org/springframework/batch/extensions/bigquery/writer/package-info.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,7 +16,7 @@
 
 /**
  * Google BigQuery related functionality.
- *
+ * <p>
  * These writers use java client from Google, so we cannot control this flow fully.
  * Take into account that this writer produces {@link com.google.cloud.bigquery.JobConfiguration.Type#LOAD} {@link com.google.cloud.bigquery.Job}.
  *
@@ -28,17 +28,17 @@
  *
  * <p>For example if you generate {@link com.google.cloud.bigquery.TableDataWriteChannel} and you {@link com.google.cloud.bigquery.TableDataWriteChannel#close()} it,
  * there is no guarantee that single {@link com.google.cloud.bigquery.Job} will be created.
- *
+ * <p>
  * Take into account that BigQuery has rate limits, and it is very easy to exceed those in concurrent environment.
- * @see <a href="https://cloud.google.com/bigquery/quotas">BigQuery Quotas &amp; Limits</a>
- *
- * Also worth mentioning that you should ensure ordering of the fields in DTO that you are going to send to the BigQuery.
+ * <p>
+ * Also, worth mentioning that you should ensure ordering of the fields in DTO that you are going to send to the BigQuery.
  * In case of CSV/JSON and Jackson consider using {@link com.fasterxml.jackson.annotation.JsonPropertyOrder}.
  *
  * @author Volodymyr Perebykivskyi
- * @since 0.1.0
+ * @since 0.2.0
  * @see <a href="https://cloud.google.com/bigquery/">Google BigQuery</a>
  * @see <a href="https://github.com/googleapis/java-bigquery">BigQuery Java Client on GitHub</a>
+ * @see <a href="https://cloud.google.com/bigquery/quotas">BigQuery Quotas &amp; Limits</a>
  */
 @NonNullApi
 package org.springframework.batch.extensions.bigquery.writer;
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/BigQueryDataLoader.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/BigQueryDataLoader.java
index 08251d5..6a3dcaf 100644
--- a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/BigQueryDataLoader.java
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/BigQueryDataLoader.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -27,13 +27,20 @@ import org.springframework.batch.extensions.bigquery.writer.builder.BigQueryCsvI
 import org.springframework.batch.extensions.bigquery.writer.builder.BigQueryJsonItemWriterBuilder;
 import org.springframework.batch.item.Chunk;
 
+import java.util.Comparator;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
 
 public class BigQueryDataLoader {
 
-    public static final Chunk<PersonDto> CHUNK = Chunk.of(
-            new PersonDto("Volodymyr", 27), new PersonDto("Oleksandra", 26)
-    );
+    /** Order must be defined so later executed queries results could be predictable */
+    private static final List<PersonDto> PERSONS = Stream
+            .of(new PersonDto("Volodymyr", 27), new PersonDto("Oleksandra", 26))
+            .sorted(Comparator.comparing(PersonDto::name))
+            .toList();
+
+    public static final Chunk<PersonDto> CHUNK = new Chunk<>(PERSONS);
 
     private final BigQuery bigQuery;
 
@@ -41,11 +48,6 @@ public class BigQueryDataLoader {
         this.bigQuery = bigQuery;
     }
 
-
-    public void loadCsvSample() throws Exception {
-        loadCsvSample(TestConstants.PERSONS_TABLE);
-    }
-
     public void loadCsvSample(String tableName) throws Exception {
         AtomicReference<Job> job = new AtomicReference<>();
 
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/PersonDto.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/PersonDto.java
index e4eabda..2d776e3 100644
--- a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/PersonDto.java
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/PersonDto.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/TestConstants.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/TestConstants.java
index 1a9c001..2835245 100644
--- a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/TestConstants.java
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/common/TestConstants.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,11 +16,19 @@
 
 package org.springframework.batch.extensions.bigquery.common;
 
-public class TestConstants {
+import com.google.cloud.bigquery.FieldValueList;
+import org.springframework.core.convert.converter.Converter;
+
+public final class TestConstants {
 
     private TestConstants() {}
 
     public static final String DATASET = "spring_batch_extensions";
-    public static final String PERSONS_TABLE = "persons";
+    public static final String NAME = "name";
+    public static final String AGE = "age";
+
+    public static final Converter<FieldValueList, PersonDto> PERSON_MAPPER = res -> new PersonDto(
+            res.get(NAME).getStringValue(), Long.valueOf(res.get(AGE).getLongValue()).intValue()
+    );
 
 }
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/base/BaseBigQueryIntegrationTest.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/base/BaseBigQueryIntegrationTest.java
new file mode 100644
index 0000000..0a8ce8c
--- /dev/null
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/base/BaseBigQueryIntegrationTest.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.springframework.batch.extensions.bigquery.integration.base;
+
+import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.BigQueryOptions;
+import org.junit.jupiter.api.TestInfo;
+
+import java.lang.reflect.Method;
+
+public abstract class BaseBigQueryIntegrationTest {
+
+    private static final String TABLE_PATTERN = "%s_%s";
+
+    public final BigQuery bigQuery = BigQueryOptions.getDefaultInstance().getService();
+
+    protected String getTableName(TestInfo testInfo) {
+        return String.format(
+                TABLE_PATTERN,
+                testInfo.getTags().stream().findFirst().orElseThrow(),
+                testInfo.getTestMethod().map(Method::getName).orElseThrow()
+        );
+    }
+}
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/package-info.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/package-info.java
index 63d60f4..aae6410 100644
--- a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/package-info.java
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/package-info.java
@@ -1,11 +1,28 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.
+ */
+
 /**
  * In order to launch these tests you should provide a way how to authorize to Google BigQuery.
  * A simple way is to create service account, store credentials as JSON file and provide environment variable.
  * Example: GOOGLE_APPLICATION_CREDENTIALS=/home/dgray/Downloads/bq-key.json
- * @see <a href="https://cloud.google.com/bigquery/docs/quickstarts/quickstart-client-libraries#before-you-begin">Authentication</a>
- *
+ * <p>
  * Test names should follow this pattern: test1, test2, testN.
  * So later in BigQuery you will see generated table name: csv_test1, csv_test2, csv_testN.
  * This way it will be easier to trace errors in BigQuery.
+ *
+ * @see <a href="https://cloud.google.com/bigquery/docs/quickstarts/quickstart-client-libraries#before-you-begin">Authentication</a>
  */
 package org.springframework.batch.extensions.bigquery.integration;
\ No newline at end of file
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/BigQueryInteractiveQueryCsvItemReaderTest.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/BigQueryInteractiveQueryCsvItemReaderTest.java
new file mode 100644
index 0000000..3ba89cc
--- /dev/null
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/BigQueryInteractiveQueryCsvItemReaderTest.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.springframework.batch.extensions.bigquery.integration.reader;
+
+import org.apache.commons.lang3.math.NumberUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.springframework.batch.extensions.bigquery.common.BigQueryDataLoader;
+import org.springframework.batch.extensions.bigquery.common.PersonDto;
+import org.springframework.batch.extensions.bigquery.common.TestConstants;
+import org.springframework.batch.extensions.bigquery.integration.reader.base.BaseCsvJsonInteractiveQueryItemReaderTest;
+import org.springframework.batch.extensions.bigquery.reader.BigQueryInteractiveQueryItemReader;
+import org.springframework.batch.extensions.bigquery.reader.builder.BigQueryInteractiveQueryItemReaderBuilder;
+import org.springframework.batch.item.Chunk;
+
+@Tag("csv")
+public class BigQueryInteractiveQueryCsvItemReaderTest extends BaseCsvJsonInteractiveQueryItemReaderTest {
+
+    @Test
+    void interactiveQueryTest1(TestInfo testInfo) throws Exception {
+        String tableName = getTableName(testInfo);
+        new BigQueryDataLoader(bigQuery).loadCsvSample(tableName);
+        Chunk<PersonDto> chunk = BigQueryDataLoader.CHUNK;
+
+        BigQueryInteractiveQueryItemReader<PersonDto> reader = new BigQueryInteractiveQueryItemReaderBuilder<PersonDto>()
+                .bigQuery(bigQuery)
+                .query(String.format("SELECT p.name, p.age FROM spring_batch_extensions.%s p ORDER BY p.name LIMIT 2", tableName))
+                .rowMapper(TestConstants.PERSON_MAPPER)
+                .build();
+
+        reader.afterPropertiesSet();
+
+        PersonDto actualFirstPerson = reader.read();
+        PersonDto expectedFirstPerson = chunk.getItems().get(0);
+
+        PersonDto actualSecondPerson = reader.read();
+        PersonDto expectedSecondPerson = chunk.getItems().get(1);
+
+        PersonDto actualThirdPerson = reader.read();
+
+        Assertions.assertNotNull(actualFirstPerson);
+        Assertions.assertEquals(expectedFirstPerson.name(), actualFirstPerson.name());
+        Assertions.assertEquals(expectedFirstPerson.age().compareTo(actualFirstPerson.age()), NumberUtils.INTEGER_ZERO);
+
+        Assertions.assertNotNull(actualSecondPerson);
+        Assertions.assertEquals(expectedSecondPerson.name(), actualSecondPerson.name());
+        Assertions.assertEquals(expectedSecondPerson.age().compareTo(actualSecondPerson.age()), NumberUtils.INTEGER_ZERO);
+
+        Assertions.assertNull(actualThirdPerson);
+    }
+
+}
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/BigQueryInteractiveQueryJsonItemReaderTest.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/BigQueryInteractiveQueryJsonItemReaderTest.java
new file mode 100644
index 0000000..98f6cf5
--- /dev/null
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/BigQueryInteractiveQueryJsonItemReaderTest.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.springframework.batch.extensions.bigquery.integration.reader;
+
+import org.apache.commons.lang3.math.NumberUtils;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+import org.springframework.batch.extensions.bigquery.common.BigQueryDataLoader;
+import org.springframework.batch.extensions.bigquery.common.PersonDto;
+import org.springframework.batch.extensions.bigquery.common.TestConstants;
+import org.springframework.batch.extensions.bigquery.integration.reader.base.BaseCsvJsonInteractiveQueryItemReaderTest;
+import org.springframework.batch.extensions.bigquery.reader.BigQueryInteractiveQueryItemReader;
+import org.springframework.batch.extensions.bigquery.reader.builder.BigQueryInteractiveQueryItemReaderBuilder;
+import org.springframework.batch.item.Chunk;
+
+@Tag("json")
+public class BigQueryInteractiveQueryJsonItemReaderTest extends BaseCsvJsonInteractiveQueryItemReaderTest {
+
+    @Test
+    void interactiveQueryTest1(TestInfo testInfo) throws Exception {
+        String tableName = getTableName(testInfo);
+        new BigQueryDataLoader(bigQuery).loadJsonSample(tableName);
+        Chunk<PersonDto> chunk = BigQueryDataLoader.CHUNK;
+
+        BigQueryInteractiveQueryItemReader<PersonDto> reader = new BigQueryInteractiveQueryItemReaderBuilder<PersonDto>()
+                .bigQuery(bigQuery)
+                .query(String.format("SELECT p.name, p.age FROM spring_batch_extensions.%s p ORDER BY p.name LIMIT 2", tableName))
+                .rowMapper(TestConstants.PERSON_MAPPER)
+                .build();
+
+        reader.afterPropertiesSet();
+
+        PersonDto actualFirstPerson = reader.read();
+        PersonDto expectedFirstPerson = chunk.getItems().get(0);
+
+        PersonDto actualSecondPerson = reader.read();
+        PersonDto expectedSecondPerson = chunk.getItems().get(1);
+
+        PersonDto actualThirdPerson = reader.read();
+
+        Assertions.assertNotNull(actualFirstPerson);
+        Assertions.assertEquals(expectedFirstPerson.name(), actualFirstPerson.name());
+        Assertions.assertEquals(expectedFirstPerson.age().compareTo(actualFirstPerson.age()), NumberUtils.INTEGER_ZERO);
+
+        Assertions.assertNotNull(actualSecondPerson);
+        Assertions.assertEquals(expectedSecondPerson.name(), actualSecondPerson.name());
+        Assertions.assertEquals(expectedSecondPerson.age().compareTo(actualSecondPerson.age()), NumberUtils.INTEGER_ZERO);
+
+        Assertions.assertNull(actualThirdPerson);
+    }
+
+}
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/base/BaseCsvJsonInteractiveQueryItemReaderTest.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/base/BaseCsvJsonInteractiveQueryItemReaderTest.java
new file mode 100644
index 0000000..97657ed
--- /dev/null
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/reader/base/BaseCsvJsonInteractiveQueryItemReaderTest.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.springframework.batch.extensions.bigquery.integration.reader.base;
+
+import com.google.cloud.bigquery.DatasetInfo;
+import com.google.cloud.bigquery.StandardTableDefinition;
+import com.google.cloud.bigquery.TableDefinition;
+import com.google.cloud.bigquery.TableId;
+import com.google.cloud.bigquery.TableInfo;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.TestInfo;
+import org.springframework.batch.extensions.bigquery.common.PersonDto;
+import org.springframework.batch.extensions.bigquery.common.TestConstants;
+import org.springframework.batch.extensions.bigquery.integration.base.BaseBigQueryIntegrationTest;
+
+import java.util.Objects;
+
+public abstract class BaseCsvJsonInteractiveQueryItemReaderTest extends BaseBigQueryIntegrationTest {
+
+    @BeforeEach
+    void prepareTest(TestInfo testInfo) {
+        if (Objects.isNull(bigQuery.getDataset(TestConstants.DATASET))) {
+            bigQuery.create(DatasetInfo.of(TestConstants.DATASET));
+        }
+
+        String tableName = getTableName(testInfo);
+
+        if (Objects.isNull(bigQuery.getTable(TestConstants.DATASET, tableName))) {
+            TableDefinition tableDefinition = StandardTableDefinition.of(PersonDto.getBigQuerySchema());
+            bigQuery.create(TableInfo.of(TableId.of(TestConstants.DATASET, tableName), tableDefinition));
+        }
+    }
+
+    @AfterEach
+    void cleanupTest(TestInfo testInfo) {
+        bigQuery.delete(TableId.of(TestConstants.DATASET, getTableName(testInfo)));
+    }
+
+}
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/BigQueryCsvItemWriterTest.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/BigQueryCsvItemWriterTest.java
index 1b07f0d..1819b2b 100644
--- a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/BigQueryCsvItemWriterTest.java
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/BigQueryCsvItemWriterTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/BigQueryJsonItemWriterTest.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/BigQueryJsonItemWriterTest.java
index e48f67d..ed577c4 100644
--- a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/BigQueryJsonItemWriterTest.java
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/BigQueryJsonItemWriterTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/base/BaseBigQueryItemWriterTest.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/base/BaseBigQueryItemWriterTest.java
index d3b99fb..4b24f8e 100644
--- a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/base/BaseBigQueryItemWriterTest.java
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/integration/writer/base/BaseBigQueryItemWriterTest.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,33 +16,21 @@
 
 package org.springframework.batch.extensions.bigquery.integration.writer.base;
 
-import com.google.cloud.RetryOption;
-import com.google.cloud.bigquery.BigQuery;
-import com.google.cloud.bigquery.BigQueryOptions;
 import com.google.cloud.bigquery.DatasetInfo;
-import com.google.cloud.bigquery.FormatOptions;
-import com.google.cloud.bigquery.JobId;
-import com.google.cloud.bigquery.JobStatus;
 import com.google.cloud.bigquery.StandardTableDefinition;
 import com.google.cloud.bigquery.TableDefinition;
 import com.google.cloud.bigquery.TableId;
 import com.google.cloud.bigquery.TableInfo;
-import com.google.cloud.bigquery.WriteChannelConfiguration;
-import org.apache.commons.lang3.BooleanUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestInfo;
 import org.springframework.batch.extensions.bigquery.common.PersonDto;
 import org.springframework.batch.extensions.bigquery.common.TestConstants;
+import org.springframework.batch.extensions.bigquery.integration.base.BaseBigQueryIntegrationTest;
 
-import java.lang.reflect.Method;
 import java.util.Objects;
 
-public abstract class BaseBigQueryItemWriterTest {
-
-    private static final String TABLE_PATTERN = "%s_%s";
-
-    protected final BigQuery bigQuery = BigQueryOptions.getDefaultInstance().getService();
+public abstract class BaseBigQueryItemWriterTest extends BaseBigQueryIntegrationTest {
 
     @BeforeEach
     void prepareTest(TestInfo testInfo) {
@@ -61,30 +49,4 @@ public abstract class BaseBigQueryItemWriterTest {
         bigQuery.delete(TableId.of(TestConstants.DATASET, getTableName(testInfo)));
     }
 
-    protected String getTableName(TestInfo testInfo) {
-        return String.format(
-                TABLE_PATTERN,
-                testInfo.getTags().stream().findFirst().orElseThrow(),
-                testInfo.getTestMethod().map(Method::getName).orElseThrow()
-        );
-    }
-
-    protected WriteChannelConfiguration generateConfiguration(TestInfo testInfo, FormatOptions formatOptions) {
-        return WriteChannelConfiguration
-                .newBuilder(TableId.of(TestConstants.DATASET, getTableName(testInfo)))
-                .setSchema(PersonDto.getBigQuerySchema())
-                .setAutodetect(false)
-                .setFormatOptions(formatOptions)
-                .build();
-    }
-
-    /** TODO check {@link com.google.cloud.bigquery.Job#waitFor(RetryOption...)} */
-    protected void waitForJobToFinish(JobId jobId) {
-        JobStatus status = bigQuery.getJob(jobId).getStatus();
-
-        while (BooleanUtils.isFalse(JobStatus.State.DONE.equals(status.getState()))) {
-            status = bigQuery.getJob(jobId).getStatus();
-        }
-    }
-
 }
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/base/AbstractBigQueryTest.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/base/AbstractBigQueryTest.java
index effc743..018a535 100644
--- a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/base/AbstractBigQueryTest.java
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/base/AbstractBigQueryTest.java
@@ -1,3 +1,19 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.springframework.batch.extensions.bigquery.unit.base;
 
 import com.google.cloud.bigquery.BigQuery;
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/reader/builder/BigQueryInteractiveQueryItemReaderBuilderTests.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/reader/builder/BigQueryInteractiveQueryItemReaderBuilderTests.java
new file mode 100644
index 0000000..dc9b4fe
--- /dev/null
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/reader/builder/BigQueryInteractiveQueryItemReaderBuilderTests.java
@@ -0,0 +1,67 @@
+/*
+ * Copyright 2002-2023 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.springframework.batch.extensions.bigquery.unit.reader.builder;
+
+import com.google.cloud.bigquery.BigQuery;
+import com.google.cloud.bigquery.QueryJobConfiguration;
+import com.google.cloud.bigquery.TableId;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.springframework.batch.extensions.bigquery.common.PersonDto;
+import org.springframework.batch.extensions.bigquery.common.TestConstants;
+import org.springframework.batch.extensions.bigquery.unit.base.AbstractBigQueryTest;
+import org.springframework.batch.extensions.bigquery.reader.BigQueryInteractiveQueryItemReader;
+import org.springframework.batch.extensions.bigquery.reader.builder.BigQueryInteractiveQueryItemReaderBuilder;
+
+class BigQueryInteractiveQueryItemReaderBuilderTests extends AbstractBigQueryTest {
+
+    @Test
+    void testSimpleQueryItemReader() {
+        BigQuery mockedBigQuery = prepareMockedBigQuery();
+
+        BigQueryInteractiveQueryItemReader<PersonDto> reader = new BigQueryInteractiveQueryItemReaderBuilder<PersonDto>()
+                .bigQuery(mockedBigQuery)
+                .query("SELECT p.name, p.age FROM spring_batch_extensions.persons p LIMIT 1")
+                .rowMapper(TestConstants.PERSON_MAPPER)
+                .build();
+
+        reader.afterPropertiesSet();
+
+        Assertions.assertNotNull(reader);
+    }
+
+    @Test
+    void testCustomQueryItemReader() {
+        BigQuery mockedBigQuery = prepareMockedBigQuery();
+
+        QueryJobConfiguration jobConfiguration = QueryJobConfiguration
+                .newBuilder("SELECT p.name, p.age FROM spring_batch_extensions.persons p LIMIT 2")
+                .setDestinationTable(TableId.of(TestConstants.DATASET, "persons_duplicate"))
+                .build();
+
+        BigQueryInteractiveQueryItemReader<PersonDto> reader = new BigQueryInteractiveQueryItemReaderBuilder<PersonDto>()
+                .bigQuery(mockedBigQuery)
+                .jobConfiguration(jobConfiguration)
+                .rowMapper(TestConstants.PERSON_MAPPER)
+                .build();
+
+        reader.afterPropertiesSet();
+
+        Assertions.assertNotNull(reader);
+    }
+
+}
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryCsvItemWriterBuilderTests.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryCsvItemWriterBuilderTests.java
index 3486a19..9b18b99 100644
--- a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryCsvItemWriterBuilderTests.java
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryCsvItemWriterBuilderTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
diff --git a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryJsonItemWriterBuilderTests.java b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryJsonItemWriterBuilderTests.java
index 3755693..0eb463e 100644
--- a/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryJsonItemWriterBuilderTests.java
+++ b/spring-batch-bigquery/src/test/java/org/springframework/batch/extensions/bigquery/unit/writer/builder/BigQueryJsonItemWriterBuilderTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -37,6 +37,8 @@ import org.springframework.batch.extensions.bigquery.writer.builder.BigQueryJson
 
 class BigQueryJsonItemWriterBuilderTests extends AbstractBigQueryTest {
 
+    private static final String TABLE = "persons_json";
+
     private final Log logger = LogFactory.getLog(getClass());
 
     /**
@@ -48,7 +50,7 @@ class BigQueryJsonItemWriterBuilderTests extends AbstractBigQueryTest {
         ObjectMapper objectMapper = new ObjectMapper();
 
         WriteChannelConfiguration writeConfiguration = WriteChannelConfiguration
-                .newBuilder(TableId.of(TestConstants.DATASET, "persons_json"))
+                .newBuilder(TableId.of(TestConstants.DATASET, TABLE))
                 .setFormatOptions(FormatOptions.json())
                 .setSchema(Schema.of(
                         Field.newBuilder("name", StandardSQLTypeName.STRING).setMode(Field.Mode.REQUIRED).build()
@@ -59,7 +61,7 @@ class BigQueryJsonItemWriterBuilderTests extends AbstractBigQueryTest {
                 .bigQuery(mockedBigQuery)
                 .rowMapper(dto -> convertDtoToJsonByteArray(objectMapper, dto))
                 .writeChannelConfig(writeConfiguration)
-                .jobConsumer(job -> this.logger.debug("Job with id: " + job.getJobId() + " is created"))
+                .jobConsumer(job -> this.logger.debug("Job with id: {}" + job.getJobId() + " is created"))
                 .build();
 
         writer.afterPropertiesSet();
@@ -72,7 +74,7 @@ class BigQueryJsonItemWriterBuilderTests extends AbstractBigQueryTest {
         BigQuery mockedBigQuery = prepareMockedBigQuery();
 
         WriteChannelConfiguration writeConfiguration = WriteChannelConfiguration
-                .newBuilder(TableId.of(TestConstants.DATASET, "persons_json"))
+                .newBuilder(TableId.of(TestConstants.DATASET, TABLE))
                 .setAutodetect(true)
                 .setFormatOptions(FormatOptions.json())
                 .build();