Skip to content

Commit

Permalink
Spark: Remove closing of IO in SerializableTable*
Browse files Browse the repository at this point in the history 
This is to fix: #12046

To summarize, the issue is that Spark can remove broadcast variables from memory and
persist them to disk in case that memory needs to be freed. In the case
that this happens, the IO object would be closed even if it was still
being used by tasks.

This fixes the issue by removing the closure of the IO object when the
serializable table is closed. The IO objects should be closed on thread
finalizers.
  • Loading branch information
@mgmarino
mgmarino committed Jan 29, 2025
1 parent e89798e commit cb81b3b
Show file tree
Hide file tree
Showing 3 changed files with 9 additions and 81 deletions.
36 changes: 3 additions & 33 deletions ...k/v3.3/spark/src/main/java/org/apache/iceberg/spark/source/SerializableTableWithSize.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,30 +22,22 @@
import org.apache.iceberg.SerializableTable;
import org.apache.iceberg.Table;
import org.apache.spark.util.KnownSizeEstimation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* This class provides a serializable table with a known size estimate. Spark calls its
* SizeEstimator class when broadcasting variables and this can be an expensive operation, so
* providing a known size estimate allows that operation to be skipped.
*
* <p>This class also implements AutoCloseable to avoid leaking resources upon broadcasting.
* Broadcast variables are destroyed and cleaned up on the driver and executors once they are
* garbage collected on the driver. The implementation ensures only resources used by copies of the
* main table are released.
* <p>This class explicitly does *not* close IO objects, because broadcast variables can be removed from memory and
* persisted to disk by Spark in the case that memory is needed, even if associated IO objects are still in use.
*/
public class SerializableTableWithSize extends SerializableTable
implements KnownSizeEstimation, AutoCloseable {
implements KnownSizeEstimation {

private static final Logger LOG = LoggerFactory.getLogger(SerializableTableWithSize.class);
private static final long SIZE_ESTIMATE = 32_768L;

private final transient Object serializationMarker;

protected SerializableTableWithSize(Table table) {
super(table);
this.serializationMarker = new Object();
}

@Override
Expand All @@ -61,38 +53,16 @@ public static Table copyOf(Table table) {
}
}

@Override
public void close() throws Exception {
if (serializationMarker == null) {
LOG.info("Releasing resources");
io().close();
}
}

public static class SerializableMetadataTableWithSize extends SerializableMetadataTable
implements KnownSizeEstimation, AutoCloseable {

private static final Logger LOG =
LoggerFactory.getLogger(SerializableMetadataTableWithSize.class);

private final transient Object serializationMarker;

protected SerializableMetadataTableWithSize(BaseMetadataTable metadataTable) {
super(metadataTable);
this.serializationMarker = new Object();
}

@Override
public long estimatedSize() {
return SIZE_ESTIMATE;
}

@Override
public void close() throws Exception {
if (serializationMarker == null) {
LOG.info("Releasing resources");
io().close();
}
}
}
}
27 changes: 3 additions & 24 deletions ...k/v3.4/spark/src/main/java/org/apache/iceberg/spark/source/SerializableTableWithSize.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,30 +23,23 @@
import org.apache.iceberg.Table;
import org.apache.iceberg.spark.SparkExecutorCache;
import org.apache.spark.util.KnownSizeEstimation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* This class provides a serializable table with a known size estimate. Spark calls its
* SizeEstimator class when broadcasting variables and this can be an expensive operation, so
* providing a known size estimate allows that operation to be skipped.
*
* <p>This class also implements AutoCloseable to avoid leaking resources upon broadcasting.
* Broadcast variables are destroyed and cleaned up on the driver and executors once they are
* garbage collected on the driver. The implementation ensures only resources used by copies of the
* main table are released.
* <p>This class also implements AutoCloseable , but does *not* close IO objects, because broadcast
* variables can be removed from memory and persisted to disk by Spark in the case that memory is
* needed, even if associated IO objects are still in use.
*/
public class SerializableTableWithSize extends SerializableTable
implements KnownSizeEstimation, AutoCloseable {

private static final Logger LOG = LoggerFactory.getLogger(SerializableTableWithSize.class);
private static final long SIZE_ESTIMATE = 32_768L;

private final transient Object serializationMarker;

protected SerializableTableWithSize(Table table) {
super(table);
this.serializationMarker = new Object();
}

@Override
Expand All @@ -64,24 +57,14 @@ public static Table copyOf(Table table) {

@Override
public void close() throws Exception {
if (serializationMarker == null) {
LOG.info("Releasing resources");
io().close();
}
invalidateCache(name());
}

public static class SerializableMetadataTableWithSize extends SerializableMetadataTable
implements KnownSizeEstimation, AutoCloseable {

private static final Logger LOG =
LoggerFactory.getLogger(SerializableMetadataTableWithSize.class);

private final transient Object serializationMarker;

protected SerializableMetadataTableWithSize(BaseMetadataTable metadataTable) {
super(metadataTable);
this.serializationMarker = new Object();
}

@Override
Expand All @@ -91,10 +74,6 @@ public long estimatedSize() {

@Override
public void close() throws Exception {
if (serializationMarker == null) {
LOG.info("Releasing resources");
io().close();
}
invalidateCache(name());
}
}
Expand Down
27 changes: 3 additions & 24 deletions ...k/v3.5/spark/src/main/java/org/apache/iceberg/spark/source/SerializableTableWithSize.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,30 +23,23 @@
import org.apache.iceberg.Table;
import org.apache.iceberg.spark.SparkExecutorCache;
import org.apache.spark.util.KnownSizeEstimation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* This class provides a serializable table with a known size estimate. Spark calls its
* SizeEstimator class when broadcasting variables and this can be an expensive operation, so
* providing a known size estimate allows that operation to be skipped.
*
* <p>This class also implements AutoCloseable to avoid leaking resources upon broadcasting.
* Broadcast variables are destroyed and cleaned up on the driver and executors once they are
* garbage collected on the driver. The implementation ensures only resources used by copies of the
* main table are released.
* <p>This class also implements AutoCloseable , but does *not* close IO objects, because broadcast
* variables can be removed from memory and persisted to disk by Spark in the case that memory is
* needed, even if associated IO objects are still in use.
*/
public class SerializableTableWithSize extends SerializableTable
implements KnownSizeEstimation, AutoCloseable {

private static final Logger LOG = LoggerFactory.getLogger(SerializableTableWithSize.class);
private static final long SIZE_ESTIMATE = 32_768L;

private final transient Object serializationMarker;

protected SerializableTableWithSize(Table table) {
super(table);
this.serializationMarker = new Object();
}

@Override
Expand All @@ -64,24 +57,14 @@ public static Table copyOf(Table table) {

@Override
public void close() throws Exception {
if (serializationMarker == null) {
LOG.info("Releasing resources");
io().close();
}
invalidateCache(name());
}

public static class SerializableMetadataTableWithSize extends SerializableMetadataTable
implements KnownSizeEstimation, AutoCloseable {

private static final Logger LOG =
LoggerFactory.getLogger(SerializableMetadataTableWithSize.class);

private final transient Object serializationMarker;

protected SerializableMetadataTableWithSize(BaseMetadataTable metadataTable) {
super(metadataTable);
this.serializationMarker = new Object();
}

@Override
Expand All @@ -91,10 +74,6 @@ public long estimatedSize() {

@Override
public void close() throws Exception {
if (serializationMarker == null) {
LOG.info("Releasing resources");
io().close();
}
invalidateCache(name());
}
}
Expand Down

0 comments on commit cb81b3b

Please sign in to comment.