Use Streams for value auto-completion

ingef · Jan 24, 2024 · 5114ac9 · 5114ac9
1 parent dd5d148
commit 5114ac9
Show file tree

Hide file tree

Showing 5 changed files with 36 additions and 21 deletions.
diff --git a/backend/src/main/java/com/bakdata/conquery/mode/StorageHandler.java b/backend/src/main/java/com/bakdata/conquery/mode/StorageHandler.java
@@ -1,12 +1,12 @@
 package com.bakdata.conquery.mode;
 
-import java.util.List;
+import java.util.stream.Stream;
 
 import com.bakdata.conquery.io.storage.NamespaceStorage;
 import com.bakdata.conquery.models.datasets.Column;
 
 public interface StorageHandler {
 
-	List<String> lookupColumnValues(NamespaceStorage namespaceStorage, Column column);
+	Stream<String> lookupColumnValues(NamespaceStorage namespaceStorage, Column column);
 
 }
diff --git a/backend/src/main/java/com/bakdata/conquery/mode/cluster/ClusterStorageHandler.java b/backend/src/main/java/com/bakdata/conquery/mode/cluster/ClusterStorageHandler.java
@@ -1,6 +1,6 @@
 package com.bakdata.conquery.mode.cluster;
 
-import java.util.List;
+import java.util.stream.Stream;
 
 import com.bakdata.conquery.io.storage.NamespaceStorage;
 import com.bakdata.conquery.mode.StorageHandler;
@@ -11,13 +11,12 @@
 public class ClusterStorageHandler implements StorageHandler {
 
 	@Override
-	public List<String> lookupColumnValues(NamespaceStorage namespaceStorage, Column column) {
+	public Stream<String> lookupColumnValues(NamespaceStorage namespaceStorage, Column column) {
 		return namespaceStorage.getAllImports().stream()
 							   .filter(imp -> imp.getTable().equals(column.getTable()))
 							   .flatMap(imp -> {
 								   final ImportColumn importColumn = imp.getColumns()[column.getPosition()];
 								   return ((StringStore) importColumn.getTypeDescription()).iterateValues();
-							   })
-							   .toList();
+							   });
 	}
 }
diff --git a/backend/src/main/java/com/bakdata/conquery/mode/local/SqlStorageHandler.java b/backend/src/main/java/com/bakdata/conquery/mode/local/SqlStorageHandler.java
@@ -1,7 +1,6 @@
 package com.bakdata.conquery.mode.local;
 
-import java.util.Collections;
-import java.util.List;
+import java.util.stream.Stream;
 
 import com.bakdata.conquery.io.storage.NamespaceStorage;
 import com.bakdata.conquery.mode.StorageHandler;
@@ -10,7 +9,6 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.jooq.Record1;
-import org.jooq.Result;
 import org.jooq.Select;
 import org.jooq.impl.DSL;
 
@@ -21,24 +19,23 @@ public class SqlStorageHandler implements StorageHandler {
 	private final SqlExecutionService sqlExecutionService;
 
 	@Override
-	public List<String> lookupColumnValues(NamespaceStorage namespaceStorage, Column column) {
+	public Stream<String> lookupColumnValues(NamespaceStorage namespaceStorage, Column column) {
 		Select<Record1<Object>> columValuesQuery = DSL.selectDistinct(DSL.field(DSL.name(column.getName())))
 													  .from(DSL.table(DSL.name(column.getTable().getName())));
 		return queryForDistinctValues(columValuesQuery);
 	}
 
-	private List<String> queryForDistinctValues(Select<Record1<Object>> columValuesQuery) {
-		Result<?> result = sqlExecutionService.execute(columValuesQuery);
+	private Stream<String> queryForDistinctValues(Select<Record1<Object>> columValuesQuery) {
 		try {
-			return result.getValues(0, String.class).stream()
-						 // the database might return null or an empty string as a distinct value
-						 .filter(value -> value != null && !value.isEmpty())
-						 .toList();
+			return sqlExecutionService.fetchStream(columValuesQuery)
+									  .map(record -> record.get(0, String.class))
+									  // the database might return null or an empty string as a distinct value
+									  .filter(value -> value != null && !value.isBlank());
 		}
 		catch (Exception e) {
-			log.error("Expecting exactly 1 column in Result when querying for distinct values of a column. Query: {}. Error: {}", columValuesQuery, e);
+			log.error("Expecting exactly 1 column in Result when querying for distinct values of a column. Query: {}.", columValuesQuery, e);
 		}
-		return Collections.emptyList();
+		return Stream.empty();
 	}
 
 }
diff --git a/backend/src/main/java/com/bakdata/conquery/models/datasets/Column.java b/backend/src/main/java/com/bakdata/conquery/models/datasets/Column.java
@@ -156,10 +156,9 @@ public TrieSearch<FrontendValue> createTrieSearch(IndexConfig config, NamespaceS
 		final TrieSearch<FrontendValue> search = new TrieSearch<>(suffixLength, config.getSearchSplitChars());
 
 		storage.getStorageHandler()
-			   .lookupColumnValues(storage, this).stream()
+			   .lookupColumnValues(storage, this)
 			   .map(value -> new FrontendValue(value, value))
 			   .onClose(() -> log.debug("DONE processing values for {}", getId()))
-
 			   .forEach(feValue -> search.addItem(feValue, FilterSearch.extractKeywords(feValue)));
 
 

diff --git a/backend/src/main/java/com/bakdata/conquery/sql/execution/SqlExecutionService.java b/backend/src/main/java/com/bakdata/conquery/sql/execution/SqlExecutionService.java
@@ -8,6 +8,7 @@
 import java.util.List;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
 
 import com.bakdata.conquery.models.error.ConqueryError;
 import com.bakdata.conquery.models.query.resultinfo.ResultInfo;
@@ -18,6 +19,7 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.jooq.DSLContext;
+import org.jooq.Record;
 import org.jooq.Result;
 import org.jooq.Select;
 import org.jooq.exception.DataAccessException;
@@ -40,7 +42,7 @@ public SqlExecutionResult execute(SqlManagedQuery sqlQuery) {
 		return result;
 	}
 
-	public Result<?> execute(Select<?> query) {
+	public Result<?> fetch(Select<?> query) {
 		log.debug("Executing query: \n{}", query);
 		try {
 			return dslContext.fetch(query);
@@ -50,6 +52,24 @@ public Result<?> execute(Select<?> query) {
 		}
 	}
 
+	/**
+	 * Executes the query and returns the results as a Stream.
+	 * <p>
+	 * Note: The returned Stream is resourceful. It must be closed by the caller, because it contains a reference to an open ResultSet (and PreparedStatement).
+	 *
+	 * @param query The query to be executed.
+	 * @return A Stream of query results.
+	 */
+	public <R extends Record> Stream<R> fetchStream(Select<R> query) {
+		log.debug("Executing query: \n{}", query);
+		try {
+			return dslContext.fetchStream(query);
+		}
+		catch (DataAccessException exception) {
+			throw new ConqueryError.SqlError(exception);
+		}
+	}
+
 	private SqlExecutionResult createStatementAndExecute(SqlManagedQuery sqlQuery, Connection connection) {
 
 		String sqlString = sqlQuery.getSqlQuery().getSql();