Improve read session caching key (#1118)

* Improve read session caching key * update changes file
GoogleCloudDataproc · Nov 6, 2023 · 00b0c4b · 00b0c4b
1 parent 2bd1df6
commit 00b0c4b
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 7 deletions.
diff --git a/CHANGES.md b/CHANGES.md
@@ -3,6 +3,7 @@
 ## Next
 
 * PR #1117: Make read session caching duration configurable
+* PR #1118: Improve read session caching key
 
 ## 0.34.0 - 2023-10-31
 

diff --git a/...main/java/com/google/cloud/spark/bigquery/v2/context/BigQueryDataSourceReaderContext.java b/...main/java/com/google/cloud/spark/bigquery/v2/context/BigQueryDataSourceReaderContext.java
@@ -43,7 +43,7 @@
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Streams;
 import java.util.ArrayList;
-import java.util.Arrays;
+import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -98,7 +98,7 @@ public OptionalLong numRows() {
  private final String applicationId;
  private Optional<StructType> schema;
  private Optional<StructType> userProvidedSchema;
- private Filter[] pushedFilters = new Filter[] {};
+ private final Set<Filter> pushedFilters = new HashSet<>();
  private Filter[] allFilters = new Filter[] {};
  private Map<String, StructField> fields;
  private ImmutableList<String> selectedFields;
@@ -199,7 +199,7 @@ public Optional<String> getCombinedFilter() {
  readSessionCreatorConfig.getPushAllFilters(),
  readSessionCreatorConfig.getReadDataFormat(),
  globalFilter,
- pushedFilters));
+ pushedFilters.toArray(new Filter[0])));
  }
 
  public Stream<InputPartitionContext<ColumnarBatch>> planBatchInputPartitionContexts() {
@@ -330,12 +330,12 @@ public Filter[] pushFilters(Filter[] filters) {
  }
 
  allFilters = filters;
- pushedFilters = handledFilters.stream().toArray(Filter[]::new);
+ pushedFilters.addAll(handledFilters);
  return unhandledFilters.stream().toArray(Filter[]::new);
  }
 
  public Filter[] pushedFilters() {
- return pushedFilters;
+ return pushedFilters.toArray(new Filter[0]);
  }
 
  public Filter[] getAllFilters() {
@@ -368,8 +368,7 @@ public Optional<List<ArrowInputPartitionContext>> filter(Filter[] filters) {
  BigQueryUtil.friendlyTableName(tableId));
  return Optional.empty();
  }
- pushedFilters =
- Stream.concat(Arrays.stream(pushedFilters), newFilters.stream()).toArray(Filter[]::new);
+ pushedFilters.addAll(newFilters);
  Optional<String> combinedFilter = getCombinedFilter();
  if (!BigQueryUtil.filterLengthInLimit(combinedFilter)) {
  logger.warn(