You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by et...@apache.org on 2025/08/22 04:46:18 UTC
(iceberg) branch main updated: Spark: Fix errorprone warnings (#13896)
This is an automated email from the ASF dual-hosted git repository.
etudenhoefner pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 36bb82675f Spark: Fix errorprone warnings (#13896)
36bb82675f is described below
commit 36bb82675ff68ac0ed059d4db62550d30aa35760
Author: Yuya Ebihara <eb...@gmail.com>
AuthorDate: Fri Aug 22 13:46:12 2025 +0900
Spark: Fix errorprone warnings (#13896)
---
.../org/apache/iceberg/spark/SparkCatalog.java | 3 +-
.../org/apache/iceberg/spark/SparkTableUtil.java | 5 ++-
.../apache/iceberg/spark/SupportsFunctions.java | 2 +
.../actions/DeleteOrphanFilesSparkAction.java | 8 +---
.../actions/RemoveDanglingDeletesSparkAction.java | 1 +
.../spark/actions/RewriteDataFilesSparkAction.java | 4 ++
.../RewritePositionDeleteFilesSparkAction.java | 3 ++
.../spark/actions/RewriteTablePathSparkAction.java | 1 +
.../actions/SparkShufflingFileRewriteRunner.java | 2 +-
.../iceberg/spark/data/SparkPlannedAvroReader.java | 1 +
.../data/vectorized/IcebergArrowColumnVector.java | 1 +
.../data/vectorized/VectorizedSparkOrcReaders.java | 11 +----
.../iceberg/spark/functions/TruncateFunction.java | 3 +-
.../procedures/ComputePartitionStatsProcedure.java | 1 +
.../procedures/CreateChangelogViewProcedure.java | 2 +-
.../procedures/FastForwardBranchProcedure.java | 1 +
.../spark/procedures/PublishChangesProcedure.java | 2 +-
.../procedures/SetCurrentSnapshotProcedure.java | 2 +-
.../iceberg/spark/source/SparkChangelogScan.java | 8 +++-
.../iceberg/spark/source/SparkCleanupUtil.java | 5 ++-
.../iceberg/spark/source/SparkCopyOnWriteScan.java | 1 +
.../spark/source/SparkMicroBatchStream.java | 3 +-
.../spark/source/SparkPartitioningAwareScan.java | 2 +-
.../spark/source/SparkPositionDeltaWrite.java | 6 ++-
.../iceberg/spark/source/SparkScanBuilder.java | 3 +-
.../iceberg/spark/source/SparkStagedScan.java | 6 +--
.../apache/iceberg/spark/source/SparkWrite.java | 49 +++++++++++++++-------
.../iceberg/spark/source/StreamingOffset.java | 6 ++-
.../iceberg/spark/source/StructInternalRow.java | 13 +++---
.../iceberg/spark/source/metrics/NumDeletes.java | 3 +-
.../iceberg/spark/source/metrics/NumSplits.java | 3 +-
31 files changed, 104 insertions(+), 57 deletions(-)
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
index 9a9ff0e00e..208ec52c8f 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkCatalog.java
@@ -191,7 +191,8 @@ public class SparkCatalog extends BaseCatalog {
SnapshotRef ref = sparkTable.table().refs().get(version);
ValidationException.check(
ref != null,
- "Cannot find matching snapshot ID or reference name for version " + version);
+ "Cannot find matching snapshot ID or reference name for version %s",
+ version);
if (ref.isBranch()) {
return sparkTable.copyWithBranch(version);
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java
index 4f585a234e..18205e3361 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java
@@ -371,8 +371,11 @@ public class SparkTableUtil {
TaskContext ctx = TaskContext.get();
String suffix =
String.format(
+ Locale.ROOT,
"stage-%d-task-%d-manifest-%s",
- ctx.stageId(), ctx.taskAttemptId(), UUID.randomUUID());
+ ctx.stageId(),
+ ctx.taskAttemptId(),
+ UUID.randomUUID());
Path location = new Path(basePath, suffix);
String outputPath = FileFormat.AVRO.addExtension(location.toString());
OutputFile outputFile = io.newOutputFile(outputPath);
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SupportsFunctions.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SupportsFunctions.java
index 34897d2b4c..83e56d1bd9 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SupportsFunctions.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/SupportsFunctions.java
@@ -35,6 +35,7 @@ interface SupportsFunctions extends FunctionCatalog {
return namespace.length == 0;
}
+ @Override
default Identifier[] listFunctions(String[] namespace) throws NoSuchNamespaceException {
if (isFunctionNamespace(namespace)) {
return SparkFunctions.list().stream()
@@ -47,6 +48,7 @@ interface SupportsFunctions extends FunctionCatalog {
throw new NoSuchNamespaceException(namespace);
}
+ @Override
default UnboundFunction loadFunction(Identifier ident) throws NoSuchFunctionException {
String[] namespace = ident.namespace();
String name = ident.name();
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java
index 09ca531ac5..46bd67c9b2 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/DeleteOrphanFilesSparkAction.java
@@ -46,7 +46,6 @@ import org.apache.iceberg.io.SupportsBulkOperations;
import org.apache.iceberg.io.SupportsPrefixOperations;
import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
-import org.apache.iceberg.relocated.com.google.common.base.Strings;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
@@ -241,7 +240,8 @@ public class DeleteOrphanFilesSparkAction extends BaseSparkAction<DeleteOrphanFi
LOG.info("Deleted {} files using bulk deletes", paths.size());
} catch (BulkDeletionFailureException e) {
int deletedFilesCount = paths.size() - e.numberFailedObjects();
- LOG.warn("Deleted only {} of {} files using bulk deletes", deletedFilesCount, paths.size());
+ LOG.warn(
+ "Deleted only {} of {} files using bulk deletes", deletedFilesCount, paths.size(), e);
}
}
@@ -488,10 +488,6 @@ public class DeleteOrphanFilesSparkAction extends BaseSparkAction<DeleteOrphanFi
return null;
}
}
-
- private boolean uriComponentMatch(String valid, String actual) {
- return Strings.isNullOrEmpty(valid) || valid.equalsIgnoreCase(actual);
- }
}
@VisibleForTesting
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java
index af8114fc23..bbe2847a92 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RemoveDanglingDeletesSparkAction.java
@@ -75,6 +75,7 @@ class RemoveDanglingDeletesSparkAction
return this;
}
+ @Override
public Result execute() {
if (table.specs().size() == 1 && table.spec().isUnpartitioned()) {
// ManifestFilterManager already performs this table-wide delete on each commit
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
index 4be9a64446..61d1fdfe60 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteDataFilesSparkAction.java
@@ -22,6 +22,7 @@ import java.math.RoundingMode;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
+import java.util.Locale;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
@@ -341,6 +342,7 @@ public class RewriteDataFilesSparkAction
} else if (failedCommits > maxFailedCommits) {
String errorMessage =
String.format(
+ Locale.ROOT,
"%s is true but %d rewrite commits failed. This is more than the maximum allowed failures of %d. "
+ "Check the logs to determine why the individual commits failed. If this is persistent it may help to "
+ "increase %s which will split the rewrite operation into smaller commits.",
@@ -422,6 +424,7 @@ public class RewriteDataFilesSparkAction
StructLike partition = group.info().partition();
if (partition.size() > 0) {
return String.format(
+ Locale.ROOT,
"Rewriting %d files (%s, file group %d/%d, %s (%d/%d)) in %s",
group.rewrittenFiles().size(),
runner.description(),
@@ -433,6 +436,7 @@ public class RewriteDataFilesSparkAction
table.name());
} else {
return String.format(
+ Locale.ROOT,
"Rewriting %d files (%s, file group %d/%d) in %s",
group.rewrittenFiles().size(),
runner.description(),
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java
index 6c019aa31c..ca9d562194 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewritePositionDeleteFilesSparkAction.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.spark.actions;
import java.math.RoundingMode;
import java.util.List;
+import java.util.Locale;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
@@ -355,6 +356,7 @@ public class RewritePositionDeleteFilesSparkAction
StructLike partition = group.info().partition();
if (partition.size() > 0) {
return String.format(
+ Locale.ROOT,
"Rewriting %d position delete files (%s, file group %d/%d, %s (%d/%d)) in %s",
group.rewrittenDeleteFiles().size(),
runner.description(),
@@ -366,6 +368,7 @@ public class RewritePositionDeleteFilesSparkAction
table.name());
} else {
return String.format(
+ Locale.ROOT,
"Rewriting %d position files (%s, file group %d/%d) in %s",
group.rewrittenDeleteFiles().size(),
runner.description(),
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java
index e16a3c5067..771b7a45b1 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java
@@ -475,6 +475,7 @@ public class RewriteTablePathSparkAction extends BaseSparkAction<RewriteTablePat
}
public static class RewriteContentFileResult extends RewriteResult<ContentFile<?>> {
+ @Override
public RewriteContentFileResult append(RewriteResult<ContentFile<?>> r1) {
this.copyPlan().addAll(r1.copyPlan());
this.toRewrite().addAll(r1.toRewrite());
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingFileRewriteRunner.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingFileRewriteRunner.java
index cbd3b6b59b..569eb252cb 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingFileRewriteRunner.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/SparkShufflingFileRewriteRunner.java
@@ -134,7 +134,7 @@ abstract class SparkShufflingFileRewriteRunner extends SparkDataFileRewriteRunne
List<FileScanTask> group, PartitionSpec outputSpec, int expectedOutputFiles) {
SortOrder[] ordering = Spark3Util.toOrdering(outputSortOrder(group, outputSpec));
int numShufflePartitions = Math.max(1, expectedOutputFiles * numShufflePartitionsPerFile);
- return (df) -> transformPlan(df, plan -> sortPlan(plan, ordering, numShufflePartitions));
+ return df -> transformPlan(df, plan -> sortPlan(plan, ordering, numShufflePartitions));
}
private LogicalPlan sortPlan(LogicalPlan plan, SortOrder[] ordering, int numShufflePartitions) {
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/SparkPlannedAvroReader.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/SparkPlannedAvroReader.java
index 4d9ee7f877..596f94cd05 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/SparkPlannedAvroReader.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/SparkPlannedAvroReader.java
@@ -126,6 +126,7 @@ public class SparkPlannedAvroReader implements DatumReader<InternalRow>, Support
return SparkValueReaders.map(SparkValueReaders.strings(), valueReader);
}
+ @Override
public ValueReader<?> variant(
Type partner, ValueReader<?> metadataReader, ValueReader<?> valueReader) {
return SparkValueReaders.variants();
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/IcebergArrowColumnVector.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/IcebergArrowColumnVector.java
index dc118aebe3..bf6de0e139 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/IcebergArrowColumnVector.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/IcebergArrowColumnVector.java
@@ -59,6 +59,7 @@ public class IcebergArrowColumnVector extends ColumnVector {
accessor.close();
}
+ @Override
public void closeIfFreeable() {
// If a column vector is writable or constant, it should override this method and do nothing.
// See more details at SPARK-50235, SPARK-50463 (Fixed in Spark 3.5.4)
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/VectorizedSparkOrcReaders.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/VectorizedSparkOrcReaders.java
index c030311232..76e0ee3811 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/VectorizedSparkOrcReaders.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/VectorizedSparkOrcReaders.java
@@ -165,13 +165,7 @@ public class VectorizedSparkOrcReaders {
}
return (columnVector, batchSize, batchOffsetInFile, isSelectedInUse, selected) ->
new PrimitiveOrcColumnVector(
- iPrimitive,
- batchSize,
- columnVector,
- primitiveValueReader,
- batchOffsetInFile,
- isSelectedInUse,
- selected);
+ iPrimitive, batchSize, columnVector, primitiveValueReader, isSelectedInUse, selected);
}
}
@@ -310,20 +304,17 @@ public class VectorizedSparkOrcReaders {
private static class PrimitiveOrcColumnVector extends BaseOrcColumnVector {
private final org.apache.orc.storage.ql.exec.vector.ColumnVector vector;
private final OrcValueReader<?> primitiveValueReader;
- private final long batchOffsetInFile;
PrimitiveOrcColumnVector(
Type type,
int batchSize,
org.apache.orc.storage.ql.exec.vector.ColumnVector vector,
OrcValueReader<?> primitiveValueReader,
- long batchOffsetInFile,
boolean isSelectedInUse,
int[] selected) {
super(type, batchSize, vector, isSelectedInUse, selected);
this.vector = vector;
this.primitiveValueReader = primitiveValueReader;
- this.batchOffsetInFile = batchOffsetInFile;
}
@Override
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/functions/TruncateFunction.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/functions/TruncateFunction.java
index fac90c9efe..662033718f 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/functions/TruncateFunction.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/functions/TruncateFunction.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.spark.functions;
import java.math.BigInteger;
import java.nio.ByteBuffer;
+import java.util.Locale;
import java.util.Set;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.util.BinaryUtil;
@@ -339,7 +340,7 @@ public class TruncateFunction implements UnboundFunction {
@Override
public String canonicalName() {
- return String.format("iceberg.truncate(decimal(%d,%d))", precision, scale);
+ return String.format(Locale.ROOT, "iceberg.truncate(decimal(%d,%d))", precision, scale);
}
@Override
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/ComputePartitionStatsProcedure.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/ComputePartitionStatsProcedure.java
index 08c87013b2..c81157496e 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/ComputePartitionStatsProcedure.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/ComputePartitionStatsProcedure.java
@@ -82,6 +82,7 @@ public class ComputePartitionStatsProcedure extends BaseProcedure {
return this;
}
+ @Override
public ProcedureParameter[] parameters() {
return PARAMETERS;
}
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java
index 3ab334595e..f81ddb8374 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java
@@ -292,7 +292,7 @@ public class CreateChangelogViewProcedure extends BaseProcedure {
* @return A name that can be safely used within Spark to reference a column by its name.
*/
private static String delimitedName(String columnName) {
- var delimited = columnName.startsWith("`") && columnName.endsWith("`");
+ boolean delimited = columnName.startsWith("`") && columnName.endsWith("`");
if (delimited) {
return columnName;
} else {
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/FastForwardBranchProcedure.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/FastForwardBranchProcedure.java
index c67124ec48..367f7d2a4d 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/FastForwardBranchProcedure.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/FastForwardBranchProcedure.java
@@ -68,6 +68,7 @@ public class FastForwardBranchProcedure extends BaseProcedure {
return this;
}
+ @Override
public ProcedureParameter[] parameters() {
return PARAMETERS;
}
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/PublishChangesProcedure.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/PublishChangesProcedure.java
index 76cd420b22..872972e22e 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/PublishChangesProcedure.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/PublishChangesProcedure.java
@@ -100,7 +100,7 @@ class PublishChangesProcedure extends BaseProcedure {
snapshot -> wapId.equals(WapUtil.stagedWapId(snapshot)),
null));
if (!wapSnapshot.isPresent()) {
- throw new ValidationException(String.format("Cannot apply unknown WAP ID '%s'", wapId));
+ throw new ValidationException("Cannot apply unknown WAP ID '%s'", wapId);
}
long wapSnapshotId = wapSnapshot.get().snapshotId();
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/SetCurrentSnapshotProcedure.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/SetCurrentSnapshotProcedure.java
index a6e67fa734..edd13fc095 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/SetCurrentSnapshotProcedure.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/SetCurrentSnapshotProcedure.java
@@ -120,7 +120,7 @@ class SetCurrentSnapshotProcedure extends BaseProcedure {
private long toSnapshotId(Table table, String refName) {
SnapshotRef ref = table.refs().get(refName);
- ValidationException.check(ref != null, "Cannot find matching snapshot ID for ref " + refName);
+ ValidationException.check(ref != null, "Cannot find matching snapshot ID for ref %s", refName);
return ref.snapshotId();
}
}
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScan.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScan.java
index 71b53d7026..55ea137ca1 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScan.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkChangelogScan.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.io.UncheckedIOException;
import java.util.Collections;
import java.util.List;
+import java.util.Locale;
import java.util.Objects;
import org.apache.iceberg.ChangelogScanTask;
import org.apache.iceberg.IncrementalChangelogScan;
@@ -127,13 +128,18 @@ class SparkChangelogScan implements Scan, SupportsReportStatistics {
@Override
public String description() {
return String.format(
+ Locale.ROOT,
"%s [fromSnapshotId=%d, toSnapshotId=%d, filters=%s]",
- table, startSnapshotId, endSnapshotId, Spark3Util.describe(filters));
+ table,
+ startSnapshotId,
+ endSnapshotId,
+ Spark3Util.describe(filters));
}
@Override
public String toString() {
return String.format(
+ Locale.ROOT,
"IcebergChangelogScan(table=%s, type=%s, fromSnapshotId=%d, toSnapshotId=%d, filters=%s)",
table,
expectedSchema.asStruct(),
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCleanupUtil.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCleanupUtil.java
index 5f34312816..93b27c566c 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCleanupUtil.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCleanupUtil.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source;
import java.util.List;
+import java.util.Locale;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.iceberg.ContentFile;
import org.apache.iceberg.exceptions.NotFoundException;
@@ -64,6 +65,7 @@ class SparkCleanupUtil {
return "unknown task";
} else {
return String.format(
+ Locale.ROOT,
"partition %d (task %d, attempt %d, stage %d.%d)",
taskContext.partitionId(),
taskContext.taskAttemptId(),
@@ -105,7 +107,8 @@ class SparkCleanupUtil {
"Deleted only {} of {} file(s) using bulk deletes ({})",
deletedFilesCount,
paths.size(),
- context);
+ context,
+ e);
}
}
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
index 7a6025b073..ee4be24618 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkCopyOnWriteScan.java
@@ -96,6 +96,7 @@ class SparkCopyOnWriteScan extends SparkPartitioningAwareScan<FileScanTask>
return estimateStatistics(snapshot);
}
+ @Override
public NamedReference[] filterAttributes() {
NamedReference file = Expressions.column(MetadataColumns.FILE_PATH.name());
return new NamedReference[] {file};
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
index aa8c98eb20..5a6d51ab31 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkMicroBatchStream.java
@@ -390,7 +390,7 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
boolean shouldContinueReading = true;
int curFilesAdded = 0;
- int curRecordCount = 0;
+ long curRecordCount = 0;
int curPos = 0;
// Note : we produce nextOffset with pos as non-inclusive
@@ -501,6 +501,7 @@ public class SparkMicroBatchStream implements MicroBatchStream, SupportsAdmissio
if (snapshot == null) {
throw new IllegalStateException(
String.format(
+ Locale.ROOT,
"Cannot load current offset at snapshot %d, the snapshot was expired or removed",
currentOffset.snapshotId()));
}
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
index 141dd4dcba..c9726518ee 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitioningAwareScan.java
@@ -177,7 +177,7 @@ abstract class SparkPartitioningAwareScan<T extends PartitionScanTask> extends S
for (ScanTask task : taskIterable) {
ValidationException.check(
taskJavaClass().isInstance(task),
- "Unsupported task type, expected a subtype of %s: %",
+ "Unsupported task type, expected a subtype of %s: %s",
taskJavaClass().getName(),
task.getClass().getName());
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java
index a5555ee887..88ca202cc9 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkPositionDeltaWrite.java
@@ -27,6 +27,7 @@ import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.function.Function;
import org.apache.iceberg.ContentFile;
@@ -260,6 +261,7 @@ class SparkPositionDeltaWrite implements DeltaWrite, RequiresDistributionAndOrde
String commitMsg =
String.format(
+ Locale.ROOT,
"position delta with %d data files, %d delete files and %d rewritten delete files"
+ "(scanSnapshotId: %d, conflictDetectionFilter: %s, isolationLevel: %s)",
addedDataFilesCount,
@@ -273,8 +275,10 @@ class SparkPositionDeltaWrite implements DeltaWrite, RequiresDistributionAndOrde
} else {
String commitMsg =
String.format(
+ Locale.ROOT,
"position delta with %d data files and %d delete files (no validation required)",
- addedDataFilesCount, addedDeleteFilesCount);
+ addedDataFilesCount,
+ addedDeleteFilesCount);
commitOperation(rowDelta, commitMsg);
}
}
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
index e4b9072155..968e6eeaaa 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkScanBuilder.java
@@ -89,7 +89,6 @@ public class SparkScanBuilder
private static final Logger LOG = LoggerFactory.getLogger(SparkScanBuilder.class);
private static final Predicate[] NO_PREDICATES = new Predicate[0];
- private StructType pushedAggregateSchema;
private Scan localScan;
private final SparkSession spark;
@@ -256,7 +255,7 @@ public class SparkScanBuilder
return false;
}
- pushedAggregateSchema =
+ StructType pushedAggregateSchema =
SparkSchemaUtil.convert(new Schema(aggregateEvaluator.resultType().fields()));
InternalRow[] pushedAggregateRows = new InternalRow[1];
StructLike structLike = aggregateEvaluator.result();
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkStagedScan.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkStagedScan.java
index fd299ade7f..d2eb4e5a56 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkStagedScan.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkStagedScan.java
@@ -78,9 +78,9 @@ class SparkStagedScan extends SparkScan {
return table().name().equals(that.table().name())
&& Objects.equals(taskSetId, that.taskSetId)
&& readSchema().equals(that.readSchema())
- && Objects.equals(splitSize, that.splitSize)
- && Objects.equals(splitLookback, that.splitLookback)
- && Objects.equals(openFileCost, that.openFileCost);
+ && splitSize == that.splitSize
+ && splitLookback == that.splitLookback
+ && openFileCost == that.openFileCost;
}
@Override
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
index 7b5129b0ab..c9a94090ef 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java
@@ -24,6 +24,7 @@ import static org.apache.iceberg.IsolationLevel.SNAPSHOT;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
+import java.util.Locale;
import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -302,7 +303,8 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
append.appendFile(file);
}
- commitOperation(append, String.format("append with %d new data files", numFiles));
+ commitOperation(
+ append, String.format(Locale.ROOT, "append with %d new data files", numFiles));
}
}
@@ -339,7 +341,8 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
commitOperation(
dynamicOverwrite,
- String.format("dynamic partition overwrite with %d new data files", numFiles));
+ String.format(
+ Locale.ROOT, "dynamic partition overwrite with %d new data files", numFiles));
}
}
@@ -376,7 +379,11 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
}
String commitMsg =
- String.format("overwrite by filter %s with %d new data files", overwriteExpr, numFiles);
+ String.format(
+ Locale.ROOT,
+ "overwrite by filter %s with %d new data files",
+ overwriteExpr,
+ numFiles);
commitOperation(overwriteFiles, commitMsg);
}
}
@@ -441,18 +448,22 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
// the scan may be null if the optimizer replaces it with an empty relation (e.g. false cond)
// no validation is needed in this case as the command does not depend on the table state
if (scan != null) {
- if (isolationLevel == SERIALIZABLE) {
- commitWithSerializableIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);
- } else if (isolationLevel == SNAPSHOT) {
- commitWithSnapshotIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);
- } else {
- throw new IllegalArgumentException("Unsupported isolation level: " + isolationLevel);
+ switch (isolationLevel) {
+ case SERIALIZABLE:
+ commitWithSerializableIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);
+ break;
+ case SNAPSHOT:
+ commitWithSnapshotIsolation(overwriteFiles, numOverwrittenFiles, numAddedFiles);
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported isolation level: " + isolationLevel);
}
} else {
commitOperation(
overwriteFiles,
- String.format("overwrite with %d new data files (no validation)", numAddedFiles));
+ String.format(
+ Locale.ROOT, "overwrite with %d new data files (no validation)", numAddedFiles));
}
}
@@ -470,8 +481,12 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
String commitMsg =
String.format(
+ Locale.ROOT,
"overwrite of %d data files with %d new data files, scanSnapshotId: %d, conflictDetectionFilter: %s",
- numOverwrittenFiles, numAddedFiles, scanSnapshotId, conflictDetectionFilter);
+ numOverwrittenFiles,
+ numAddedFiles,
+ scanSnapshotId,
+ conflictDetectionFilter);
commitOperation(overwriteFiles, commitMsg);
}
@@ -488,8 +503,10 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
String commitMsg =
String.format(
+ Locale.ROOT,
"overwrite of %d data files with %d new data files",
- numOverwrittenFiles, numAddedFiles);
+ numOverwrittenFiles,
+ numAddedFiles);
commitOperation(overwriteFiles, commitMsg);
}
}
@@ -588,7 +605,10 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
append.appendFile(file);
numFiles++;
}
- commit(append, epochId, String.format("streaming append with %d new data files", numFiles));
+ commit(
+ append,
+ epochId,
+ String.format(Locale.ROOT, "streaming append with %d new data files", numFiles));
}
}
@@ -610,7 +630,8 @@ abstract class SparkWrite implements Write, RequiresDistributionAndOrdering {
commit(
overwriteFiles,
epochId,
- String.format("streaming complete overwrite with %d new data files", numFiles));
+ String.format(
+ Locale.ROOT, "streaming complete overwrite with %d new data files", numFiles));
}
}
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/StreamingOffset.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/StreamingOffset.java
index f2088deb1e..8869fedc84 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/StreamingOffset.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/StreamingOffset.java
@@ -24,6 +24,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.StringWriter;
import java.io.UncheckedIOException;
+import java.util.Locale;
import org.apache.iceberg.relocated.com.google.common.base.Objects;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.util.JsonUtil;
@@ -134,8 +135,11 @@ class StreamingOffset extends Offset {
@Override
public String toString() {
return String.format(
+ Locale.ROOT,
"Streaming Offset[%d: position (%d) scan_all_files (%b)]",
- snapshotId, position, scanAllFiles);
+ snapshotId,
+ position,
+ scanAllFiles);
}
private static StreamingOffset fromJsonNode(JsonNode node) {
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/StructInternalRow.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/StructInternalRow.java
index 25a349b891..27fb014d8c 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/StructInternalRow.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/StructInternalRow.java
@@ -162,10 +162,10 @@ class StructInternalRow extends InternalRow {
@Override
public Decimal getDecimal(int ordinal, int precision, int scale) {
- return isNullAt(ordinal) ? null : getDecimalInternal(ordinal, precision, scale);
+ return isNullAt(ordinal) ? null : getDecimalInternal(ordinal);
}
- private Decimal getDecimalInternal(int ordinal, int precision, int scale) {
+ private Decimal getDecimalInternal(int ordinal) {
return Decimal.apply(struct.get(ordinal, BigDecimal.class));
}
@@ -205,10 +205,10 @@ class StructInternalRow extends InternalRow {
@Override
public InternalRow getStruct(int ordinal, int numFields) {
- return isNullAt(ordinal) ? null : getStructInternal(ordinal, numFields);
+ return isNullAt(ordinal) ? null : getStructInternal(ordinal);
}
- private InternalRow getStructInternal(int ordinal, int numFields) {
+ private InternalRow getStructInternal(int ordinal) {
return new StructInternalRow(
type.fields().get(ordinal).type().asStructType(), struct.get(ordinal, StructLike.class));
}
@@ -257,12 +257,11 @@ class StructInternalRow extends InternalRow {
} else if (dataType instanceof DoubleType) {
return getDouble(ordinal);
} else if (dataType instanceof DecimalType) {
- DecimalType decimalType = (DecimalType) dataType;
- return getDecimalInternal(ordinal, decimalType.precision(), decimalType.scale());
+ return getDecimalInternal(ordinal);
} else if (dataType instanceof BinaryType) {
return getBinaryInternal(ordinal);
} else if (dataType instanceof StructType) {
- return getStructInternal(ordinal, ((StructType) dataType).size());
+ return getStructInternal(ordinal);
} else if (dataType instanceof ArrayType) {
return getArrayInternal(ordinal);
} else if (dataType instanceof MapType) {
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/metrics/NumDeletes.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/metrics/NumDeletes.java
index 000499874b..eeff4286d0 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/metrics/NumDeletes.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/metrics/NumDeletes.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source.metrics;
import java.text.NumberFormat;
+import java.util.Locale;
import org.apache.spark.sql.connector.metric.CustomMetric;
public class NumDeletes implements CustomMetric {
@@ -42,6 +43,6 @@ public class NumDeletes implements CustomMetric {
sum += taskMetric;
}
- return NumberFormat.getIntegerInstance().format(sum);
+ return NumberFormat.getIntegerInstance(Locale.ROOT).format(sum);
}
}
diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/metrics/NumSplits.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/metrics/NumSplits.java
index 41d7c1e8db..f5c7a1f78c 100644
--- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/metrics/NumSplits.java
+++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/source/metrics/NumSplits.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.source.metrics;
import java.text.NumberFormat;
+import java.util.Locale;
import org.apache.spark.sql.connector.metric.CustomMetric;
public class NumSplits implements CustomMetric {
@@ -40,6 +41,6 @@ public class NumSplits implements CustomMetric {
sum += taskMetric;
}
- return NumberFormat.getIntegerInstance().format(sum);
+ return NumberFormat.getIntegerInstance(Locale.ROOT).format(sum);
}
}