Skip to content

Bringing in createBatchScanner refactor from older MR #3073

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 4 commits into
base: integration
Choose a base branch
from
Draft

Bringing in createBatchScanner refactor from older MR #3073

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
af2cee9
Bringing in createBatchScanner refactor from older MR
avgAGB Jul 28, 2025
c84d9d8
Merge remote-tracking branch 'origin/integration' into task/createBat…
avgAGB Jul 28, 2025
bbf4efe
Removing method call in ShardQueryConfiguration
avgAGB Jul 30, 2025
f69e27c
Merge remote-tracking branch 'origin/integration' into task/createBat…
avgAGB Aug 12, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions warehouse/query-core/src/main/java/datawave/query/scheduler/PushdownScheduler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,6 @@
import datawave.query.config.ShardQueryConfiguration;
import datawave.query.tables.BatchScannerSession;
import datawave.query.tables.ScannerFactory;
import datawave.query.tables.ShardQueryLogic;
import datawave.query.tables.async.ScannerChunk;
import datawave.query.tables.async.event.VisitorFunction;
import datawave.query.tables.stats.ScanSessionStats;
Expand Down Expand Up @@ -223,7 +222,7 @@ public void close() {
*/
@Override
public BatchScanner createBatchScanner(ShardQueryConfiguration config, ScannerFactory scannerFactory, QueryData qd) throws TableNotFoundException {
return ShardQueryLogic.createBatchScanner(config, scannerFactory, qd);
return scannerFactory.newScanner(config, qd);
}

@Override
Expand Down
26 changes: 26 additions & 0 deletions warehouse/query-core/src/main/java/datawave/query/tables/ScannerFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@

import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.Scanner;
import org.apache.accumulo.core.client.ScannerBase;
import org.apache.accumulo.core.client.TableNotFoundException;
Expand All @@ -25,6 +26,7 @@
import com.google.common.base.Preconditions;

import datawave.core.query.configuration.GenericQueryConfiguration;
import datawave.core.query.configuration.QueryData;
import datawave.ingest.data.config.ingest.AccumuloHelper;
import datawave.microservice.query.Query;
import datawave.mr.bulk.BulkInputFormat;
Expand Down Expand Up @@ -515,6 +517,30 @@ public ScannerBase newRfileScanner(String tableName, Set<Authorizations> auths,
}
}

public BatchScanner newScanner(ShardQueryConfiguration config, QueryData qd) throws TableNotFoundException {
final BatchScanner bs = this.newScanner(config.getShardTableName(), config.getAuthorizations(), config.getNumQueryThreads(), config.getQuery());

if (log.isTraceEnabled()) {
log.trace("Running with " + config.getAuthorizations() + " and " + config.getNumQueryThreads() + " threads: " + qd);
}

bs.setRanges(qd.getRanges());

for (IteratorSetting cfg : qd.getSettings()) {
bs.addScanIterator(cfg);
}

if (config.getTableConsistencyLevels().containsKey(config.getTableName())) {
bs.setConsistencyLevel(config.getTableConsistencyLevels().get(config.getTableName()));
}

if (config.getTableHints().containsKey(config.getTableName())) {
bs.setExecutionHints(config.getTableHints().get(config.getTableName()));
}

return bs;
}

/**
* Apply table-specific scanner configs to the provided scanner base object using the table name as the key
*
Expand Down
27 changes: 0 additions & 27 deletions warehouse/query-core/src/main/java/datawave/query/tables/ShardQueryLogic.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,6 @@
import java.util.concurrent.TimeUnit;

import org.apache.accumulo.core.client.AccumuloClient;
import org.apache.accumulo.core.client.BatchScanner;
import org.apache.accumulo.core.client.IteratorSetting;
import org.apache.accumulo.core.client.ScannerBase;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.data.Key;
Expand Down Expand Up @@ -277,31 +275,6 @@ public ShardQueryLogic(ShardQueryLogic other) {
}
}

public static BatchScanner createBatchScanner(ShardQueryConfiguration config, ScannerFactory scannerFactory, QueryData qd) throws TableNotFoundException {
final BatchScanner bs = scannerFactory.newScanner(config.getShardTableName(), config.getAuthorizations(), config.getNumQueryThreads(),
config.getQuery());

if (log.isTraceEnabled()) {
log.trace("Running with " + config.getAuthorizations() + " and " + config.getNumQueryThreads() + " threads: " + qd);
}

bs.setRanges(qd.getRanges());

for (IteratorSetting cfg : qd.getSettings()) {
bs.addScanIterator(cfg);
}

if (config.getTableConsistencyLevels().containsKey(config.getTableName())) {
bs.setConsistencyLevel(config.getTableConsistencyLevels().get(config.getTableName()));
}

if (config.getTableHints().containsKey(config.getTableName())) {
bs.setExecutionHints(config.getTableHints().get(config.getTableName()));
}

return bs;
}

@Override
public GenericQueryConfiguration initialize(AccumuloClient client, Query settings, Set<Authorizations> auths) throws Exception {
// whenever we reinitialize, ensure we have a fresh transformer
Expand Down