Core: Optimize the TableScanContext

core/src/main/java/org/apache/iceberg/TableScanContext.java

@@ -96,104 +96,39 @@ Long snapshotId() {
   }
 
   TableScanContext useSnapshotId(Long scanSnapshotId) {
-    return new TableScanContext(
-        scanSnapshotId,
-        rowFilter,
-        ignoreResiduals,
-        caseSensitive,
-        colStats,
-        projectedSchema,
-        selectedColumns,
-        options,
-        fromSnapshotId,
-        toSnapshotId,
-        planExecutor,
-        fromSnapshotInclusive,
-        metricsReporter);
+    return Builder.from(this).snapshotId(scanSnapshotId).build();
   }
 
   Expression rowFilter() {
     return rowFilter;
   }
 
   TableScanContext filterRows(Expression filter) {
-    return new TableScanContext(
-        snapshotId,
-        filter,
-        ignoreResiduals,
-        caseSensitive,
-        colStats,
-        projectedSchema,
-        selectedColumns,
-        options,
-        fromSnapshotId,
-        toSnapshotId,
-        planExecutor,
-        fromSnapshotInclusive,
-        metricsReporter);
+    return Builder.from(this).rowFilter(filter).build();
   }
 
   boolean ignoreResiduals() {
     return ignoreResiduals;
   }
 
   TableScanContext ignoreResiduals(boolean shouldIgnoreResiduals) {
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        shouldIgnoreResiduals,
-        caseSensitive,
-        colStats,
-        projectedSchema,
-        selectedColumns,
-        options,
-        fromSnapshotId,
-        toSnapshotId,
-        planExecutor,
-        fromSnapshotInclusive,
-        metricsReporter);
+    return Builder.from(this).ignoreResiduals(shouldIgnoreResiduals).build();
   }
 
   boolean caseSensitive() {
     return caseSensitive;
   }
 
   TableScanContext setCaseSensitive(boolean isCaseSensitive) {
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        ignoreResiduals,
-        isCaseSensitive,
-        colStats,
-        projectedSchema,
-        selectedColumns,
-        options,
-        fromSnapshotId,
-        toSnapshotId,
-        planExecutor,
-        fromSnapshotInclusive,
-        metricsReporter);
+    return Builder.from(this).caseSensitive(isCaseSensitive).build();
   }
 
   boolean returnColumnStats() {
     return colStats;
   }
 
   TableScanContext shouldReturnColumnStats(boolean returnColumnStats) {
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        ignoreResiduals,
-        caseSensitive,
-        returnColumnStats,
-        projectedSchema,
-        selectedColumns,
-        options,
-        fromSnapshotId,
-        toSnapshotId,
-        planExecutor,
-        fromSnapshotInclusive,
-        metricsReporter);
+    return Builder.from(this).colStats(returnColumnStats).build();
   }
 
   Collection<String> selectedColumns() {
@@ -203,20 +138,7 @@ Collection<String> selectedColumns() {
   TableScanContext selectColumns(Collection<String> columns) {
     Preconditions.checkState(
         projectedSchema == null, "Cannot select columns when projection schema is set");
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        ignoreResiduals,
-        caseSensitive,
-        colStats,
-        null,
-        columns,
-        options,
-        fromSnapshotId,
-        toSnapshotId,
-        planExecutor,
-        fromSnapshotInclusive,
-        metricsReporter);
+    return Builder.from(this).projectedSchema(null).selectedColumns(columns).build();
   }
 
   Schema projectedSchema() {
@@ -226,20 +148,7 @@ Schema projectedSchema() {
   TableScanContext project(Schema schema) {
     Preconditions.checkState(
         selectedColumns == null, "Cannot set projection schema when columns are selected");
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        ignoreResiduals,
-        caseSensitive,
-        colStats,
-        schema,
-        null,
-        options,
-        fromSnapshotId,
-        toSnapshotId,
-        planExecutor,
-        fromSnapshotInclusive,
-        metricsReporter);
+    return Builder.from(this).projectedSchema(schema).selectedColumns(null).build();
   }
 
   Map<String, String> options() {
@@ -250,58 +159,20 @@ TableScanContext withOption(String property, String value) {
     ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
     builder.putAll(options);
     builder.put(property, value);
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        ignoreResiduals,
-        caseSensitive,
-        colStats,
-        projectedSchema,
-        selectedColumns,
-        builder.build(),
-        fromSnapshotId,
-        toSnapshotId,
-        planExecutor,
-        fromSnapshotInclusive,
-        metricsReporter);
+
+    return Builder.from(this).options(builder.build()).build();
   }
 
   Long fromSnapshotId() {
     return fromSnapshotId;
   }
 
   TableScanContext fromSnapshotIdExclusive(long id) {
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        ignoreResiduals,
-        caseSensitive,
-        colStats,
-        projectedSchema,
-        selectedColumns,
-        options,
-        id,
-        toSnapshotId,
-        planExecutor,
-        false,
-        metricsReporter);
+    return Builder.from(this).fromSnapshotId(id).fromSnapshotInclusive(false).build();
   }
 
   TableScanContext fromSnapshotIdInclusive(long id) {
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        ignoreResiduals,
-        caseSensitive,
-        colStats,
-        projectedSchema,
-        selectedColumns,
-        options,
-        id,
-        toSnapshotId,
-        planExecutor,
-        true,
-        metricsReporter);
+    return Builder.from(this).fromSnapshotId(id).fromSnapshotInclusive(true).build();
   }
 
   boolean fromSnapshotInclusive() {
@@ -313,20 +184,7 @@ Long toSnapshotId() {
   }
 
   TableScanContext toSnapshotId(long id) {
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        ignoreResiduals,
-        caseSensitive,
-        colStats,
-        projectedSchema,
-        selectedColumns,
-        options,
-        fromSnapshotId,
-        id,
-        planExecutor,
-        fromSnapshotInclusive,
-        metricsReporter);
+    return Builder.from(this).toSnapshotId(id).build();
   }
 
   ExecutorService planExecutor() {
@@ -338,40 +196,132 @@ boolean planWithCustomizedExecutor() {
   }
 
   TableScanContext planWith(ExecutorService executor) {
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        ignoreResiduals,
-        caseSensitive,
-        colStats,
-        projectedSchema,
-        selectedColumns,
-        options,
-        fromSnapshotId,
-        toSnapshotId,
-        executor,
-        fromSnapshotInclusive,
-        metricsReporter);
+    return Builder.from(this).planExecutor(executor).build();
   }
 
   MetricsReporter metricsReporter() {
     return metricsReporter;
   }
 
   TableScanContext reportWith(MetricsReporter reporter) {
-    return new TableScanContext(
-        snapshotId,
-        rowFilter,
-        ignoreResiduals,
-        caseSensitive,
-        colStats,
-        projectedSchema,
-        selectedColumns,
-        options,
-        fromSnapshotId,
-        toSnapshotId,
-        planExecutor,
-        fromSnapshotInclusive,
-        reporter);
+    return Builder.from(this).metricsReporter(reporter).build();
+  }
+
+  private static final class Builder {
+    private Long snapshotId;
+    private Expression rowFilter;
+    private boolean ignoreResiduals;
+    private boolean caseSensitive;
+    private boolean colStats;
+    private Schema projectedSchema;
+    private Collection<String> selectedColumns;
+    private ImmutableMap<String, String> options;
+    private Long fromSnapshotId;
+    private Long toSnapshotId;
+    private ExecutorService planExecutor;
+    private boolean fromSnapshotInclusive;
+    private MetricsReporter metricsReporter;
+
+    Builder(TableScanContext context) {
+      this.snapshotId(context.snapshotId)
+          .rowFilter(context.rowFilter)
+          .ignoreResiduals(context.ignoreResiduals)
+          .caseSensitive(context.caseSensitive)
+          .colStats(context.colStats)
+          .projectedSchema(context.projectedSchema)
+          .selectedColumns(context.selectedColumns)
+          .options(context.options)
+          .fromSnapshotId(context.fromSnapshotId)
+          .toSnapshotId(context.toSnapshotId)
+          .planExecutor(context.planExecutor)
+          .fromSnapshotInclusive(context.fromSnapshotInclusive)
+          .metricsReporter(context.metricsReporter);
+    }
+
+    static Builder from(TableScanContext context) {
+      return new Builder(context);
+    }
+
+    Builder snapshotId(Long newSnapshotId) {
+      this.snapshotId = newSnapshotId;
+      return this;
+    }
+
+    Builder rowFilter(Expression newRowFilter) {
+      this.rowFilter = newRowFilter;
+      return this;
+    }
+
+    Builder ignoreResiduals(boolean newIgnoreResiduals) {
+      this.ignoreResiduals = newIgnoreResiduals;
+      return this;
+    }
+
+    Builder caseSensitive(boolean newCaseSensitive) {
+      this.caseSensitive = newCaseSensitive;
+      return this;
+    }
+
+    Builder colStats(boolean newColStats) {
+      this.colStats = newColStats;
+      return this;
+    }
+
+    Builder projectedSchema(Schema newProjectedSchema) {
+      this.projectedSchema = newProjectedSchema;
+      return this;
+    }
+
+    Builder selectedColumns(Collection<String> newSelectedColumns) {
+      this.selectedColumns = newSelectedColumns;
+      return this;
+    }
+
+    Builder options(ImmutableMap<String, String> newOptions) {
+      this.options = newOptions;
+      return this;
+    }
+
+    Builder fromSnapshotId(Long newFromSnapshotId) {
+      this.fromSnapshotId = newFromSnapshotId;
+      return this;
+    }
+
+    Builder toSnapshotId(Long newToSnapshotId) {
+      this.toSnapshotId = newToSnapshotId;
+      return this;
+    }
+
+    Builder planExecutor(ExecutorService newPlanExecutor) {
+      this.planExecutor = newPlanExecutor;
+      return this;
+    }
+
+    Builder fromSnapshotInclusive(boolean newFromSnapshotInclusive) {
+      this.fromSnapshotInclusive = newFromSnapshotInclusive;
+      return this;
+    }
+
+    Builder metricsReporter(MetricsReporter newMetricsReporter) {
+      this.metricsReporter = newMetricsReporter;
+      return this;
+    }
+
+    TableScanContext build() {
+      return new TableScanContext(
+          snapshotId,
+          rowFilter,
+          ignoreResiduals,
+          caseSensitive,
+          colStats,
+          projectedSchema,
+          selectedColumns,
+          options,
+          fromSnapshotId,
+          toSnapshotId,
+          planExecutor,
+          fromSnapshotInclusive,
+          metricsReporter);
+    }
   }
 }