Merge remote-tracking branch 'upstream/main' into refine_Spark_3_words_from_docs

Author: jackylee-ch

.baseline/checkstyle/checkstyle-suppressions.xml

@@ -51,6 +51,7 @@
 
     <!-- Allow using Flink's shaded Curator dependency -->
     <suppress files="org.apache.iceberg.flink.maintenance.api.ZkLockFactory" id="BanShadedClasses"/>
+    <suppress files="org.apache.iceberg.flink.maintenance.api.TestZkLockFactory" id="BanShadedClasses"/>
 
     <!-- Suppress checks for CometColumnReader -->
     <suppress files="org.apache.iceberg.spark.data.vectorized.CometColumnReader" checks="IllegalImport"/>

.baseline/checkstyle/checkstyle.xml

@@ -265,6 +265,12 @@
             <property name="illegalClasses" value="org.apache.flink.util.Preconditions"/>
             <message key="import.illegal" value="Use org.apache.iceberg.relocated.com.google.common.base.Preconditions instead."/>
         </module>
+        <module name="IllegalImport">
+            <property name="id" value="BanHadoopUtils"/>
+            <property name="regexp" value="true"/>
+            <property name="illegalClasses" value="^org\.apache\.hadoop\.util\.(Sets|Lists|Preconditions)"/>
+            <message key="import.illegal" value="Use org.apache.iceberg.relocated.* classes from bundled-guava module instead."/>
+        </module>
         <module name="IllegalImport">
             <property name="id" value="GuavaClassesInAssertJ"/>
             <property name="illegalPkgs" value="org.assertj.core.util"/>

README.md

@@ -71,8 +71,8 @@ Iceberg table support is organized in library modules:
 
 Iceberg also has modules for adding Iceberg support to processing engines:
 
-* `iceberg-spark` is an implementation of Spark's Datasource V2 API for Iceberg with submodules for each spark versions (use runtime jars for a shaded version)
-* `iceberg-flink` contains classes for integrating with Apache Flink (use iceberg-flink-runtime for a shaded version)
+* `iceberg-spark` is an implementation of Spark's Datasource V2 API for Iceberg with submodules for each spark versions (use [runtime jars](https://iceberg.apache.org/multi-engine-support/#runtime-jar) for a shaded version to avoid dependency conflicts)
+* `iceberg-flink` contains classes for integrating with Apache Flink (use [iceberg-flink-runtime](https://iceberg.apache.org/multi-engine-support/#runtime-jar) for a shaded version)
 * `iceberg-mr` contains an InputFormat and other classes for integrating with Apache Hive
 
 ---

api/src/main/java/org/apache/iceberg/types/Conversions.java

@@ -32,6 +32,9 @@
 import org.apache.iceberg.exceptions.RuntimeIOException;
 import org.apache.iceberg.expressions.Literal;
 import org.apache.iceberg.util.UUIDUtil;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantValue;
 
 public class Conversions {
 
@@ -117,6 +120,17 @@ public static ByteBuffer toByteBuffer(Type.TypeID typeId, Object value) {
         return (ByteBuffer) value;
       case DECIMAL:
         return ByteBuffer.wrap(((BigDecimal) value).unscaledValue().toByteArray());
+      case VARIANT:
+        // Produce a concatenated buffer of metadata and value
+        Variant variant = (Variant) value;
+        VariantMetadata variantMetadata = variant.metadata();
+        VariantValue variantValue = variant.value();
+        ByteBuffer variantBuffer =
+            ByteBuffer.allocate(variantMetadata.sizeInBytes() + variantValue.sizeInBytes())
+                .order(ByteOrder.LITTLE_ENDIAN);
+        variantMetadata.writeTo(variantBuffer, 0);
+        variantValue.writeTo(variantBuffer, variantMetadata.sizeInBytes());
+        return variantBuffer;
       default:
         throw new UnsupportedOperationException("Cannot serialize type: " + typeId);
     }
@@ -177,6 +191,8 @@ private static Object internalFromByteBuffer(Type type, ByteBuffer buffer) {
         byte[] unscaledBytes = new byte[buffer.remaining()];
         tmp.get(unscaledBytes);
         return new BigDecimal(new BigInteger(unscaledBytes), decimal.scale());
+      case VARIANT:
+        return Variant.from(tmp);
       default:
         throw new UnsupportedOperationException("Cannot deserialize type: " + type);
     }

baseline.gradle

@@ -52,15 +52,6 @@ subprojects {
   apply plugin: 'com.palantir.baseline-exact-dependencies'
   apply plugin: 'com.diffplug.spotless'
 
-  pluginManager.withPlugin('com.palantir.baseline-checkstyle') {
-    checkstyle {
-      // com.palantir.baseline:gradle-baseline-java:4.42.0 (the last version supporting Java 8) pulls
-      // in an old version of the checkstyle(9.1), which has this OutOfMemory bug https://github.com/checkstyle/checkstyle/issues/10934.
-      // So, override its checkstyle version using CheckstyleExtension to 9.3 (the latest java 8 supported version) which contains a fix.
-      toolVersion '9.3'
-    }
-  }
-
   pluginManager.withPlugin('com.diffplug.spotless') {
     spotless {
       java {
@@ -174,4 +165,22 @@ subprojects {
       quiet = false
     }
   }
+
+  pluginManager.withPlugin('scala') {
+    String scalaVersion = System.getProperty("scalaVersion") != null ? System.getProperty("scalaVersion") : System.getProperty("defaultScalaVersion")
+    tasks.withType(ScalaCompile).configureEach { scalaCompile ->
+      if (scalaVersion?.startsWith("2.12")) {
+        scalaCompile.scalaCompileOptions.additionalParameters = [
+                // Scala 2.12 does not support treating unused imports as errors individually,
+                // so we only enable warnings for unused imports when using Scala 2.12.
+                "-Ywarn-unused:imports",
+        ]
+      } else if (scalaVersion?.startsWith("2.13")) {
+        scalaCompile.scalaCompileOptions.additionalParameters = [
+                "-Wconf:cat=unused:error",
+                "-Wunused:imports"
+        ]
+      }
+    }
+  }
 }

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

@@ -300,7 +300,7 @@ private DeleteFileIndex planDeletesLocally(List<ManifestFile> deleteManifests) {
     }
 
     return builder
-        .specsById(table().specs())
+        .specsById(specs())
         .filterData(filter())
         .caseSensitive(isCaseSensitive())
         .scanMetrics(scanMetrics())

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

@@ -214,7 +214,7 @@ protected void refreshFromMetadataLocation(
     this.shouldRefresh = false;
   }
 
-  private String metadataFileLocation(TableMetadata metadata, String filename) {
+  protected String metadataFileLocation(TableMetadata metadata, String filename) {
     String metadataLocation = metadata.properties().get(TableProperties.WRITE_METADATA_LOCATION);
 
     if (metadataLocation != null) {

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

@@ -53,7 +53,7 @@ protected ManifestGroup newManifestGroup(
             .caseSensitive(isCaseSensitive())
             .select(withColumnStats ? SCAN_WITH_STATS_COLUMNS : SCAN_COLUMNS)
             .filterData(filter())
-            .specsById(table().specs())
+            .specsById(specs())
             .scanMetrics(scanMetrics())
             .ignoreDeleted()
             .columnsToKeepStats(columnsToKeepStats());

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

@@ -74,7 +74,7 @@ public CloseableIterable<FileScanTask> doPlanFiles() {
             .caseSensitive(isCaseSensitive())
             .select(scanColumns())
             .filterData(filter())
-            .specsById(table().specs())
+            .specsById(specs())
             .scanMetrics(scanMetrics())
             .ignoreDeleted()
             .columnsToKeepStats(columnsToKeepStats());

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

@@ -55,6 +55,14 @@ public final class MetricsConfig implements Serializable {
   private static final MetricsMode DEFAULT_MODE =
       MetricsModes.fromString(DEFAULT_WRITE_METRICS_MODE_DEFAULT);
   private static final MetricsConfig DEFAULT = new MetricsConfig(ImmutableMap.of(), DEFAULT_MODE);
+  private static final MetricsConfig POSITION_DELETE_MODE =
+      new MetricsConfig(
+          ImmutableMap.of(
+              MetadataColumns.DELETE_FILE_PATH.name(),
+              MetricsModes.Full.get(),
+              MetadataColumns.DELETE_FILE_POS.name(),
+              MetricsModes.Full.get()),
+          DEFAULT_MODE);
 
   private final Map<String, MetricsMode> columnModes;
   private final MetricsMode defaultMode;
@@ -68,6 +76,10 @@ public static MetricsConfig getDefault() {
     return DEFAULT;
   }
 
+  public static MetricsConfig forPositionDelete() {
+    return POSITION_DELETE_MODE;
+  }
+
   /**
    * Creates a metrics config from table configuration.
    *
@@ -92,7 +104,11 @@ public static MetricsConfig forTable(Table table) {
    * Creates a metrics config for a position delete file.
    *
    * @param table an Iceberg table
+   * @deprecated This method is deprecated as of version 1.11.0 and will be removed in 1.12.0.
+   *     Position deletes that include row data are no longer supported. Use {@link
+   *     #forPositionDelete()} instead.
    */
+  @Deprecated
   public static MetricsConfig forPositionDelete(Table table) {
     ImmutableMap.Builder<String, MetricsMode> columnModes = ImmutableMap.builder();