Initial POC - streaming table metadata rather than de/serializing in memory #1395

collado-mike · 2025-04-18T00:35:13Z

Initial POC for streaming table metadata directly from storage rather than deserializing in memory, then serializing back out. The goal is to avoid parsing very large metadata files and holding them in memory. We can avoid the need to throttle based on large file sizes and use fixed memory for handling any size TableMetadata.

Some aspects need consideration, such as reading table properties, which we currently do by parsing the TableMetadata. These properties could be written to the table entity properties in addition to the TableMetadata file so that we can look up some things from the entity directly.

Not all tests are passing yet (the catalog integration tests do). Just gauging interest.

…memory

adutra

Interesting improvement 👍

adutra · 2025-04-21T12:59:49Z

...e/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

+            .getConfiguration(
+                polarisCallContext, FeatureConfiguration.ENABLE_STREAMING_TABLE_METADATA)
+            .equals(true)) {


nit:

Suggested change

.getConfiguration(

polarisCallContext, FeatureConfiguration.ENABLE_STREAMING_TABLE_METADATA)

.equals(true)) {

.getConfiguration(

polarisCallContext, FeatureConfiguration.ENABLE_STREAMING_TABLE_METADATA)) {

adutra · 2025-04-21T13:04:07Z

...e/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

+  public static class StreamingLoadTableResponse implements RESTResponse {
+    private String metadataLocation;
+    private InputStream metadata;
+    private Map<String, String> config;
+    private List<Credential> credentials;
+
+    public StreamingLoadTableResponse() {}
+
+    public StreamingLoadTableResponse(
+        String metadataLocation,
+        InputStream metadata,
+        Map<String, String> config,
+        List<Credential> credentials) {
+      this.metadataLocation = metadataLocation;
+      this.metadata = metadata;
+      this.config = config;
+      this.credentials = credentials;
+    }
+
+    @Override
+    public void validate() {
+      Preconditions.checkNotNull(this.metadata, "Invalid metadata: null");
+    }
+
+    @JsonProperty("metadata-location")
+    public String metadataLocation() {
+      return this.metadataLocation;
+    }
+
+    @JsonProperty("metadata")
+    @JsonSerialize(using = InputStreamSerializer.class)
+    public InputStream tableMetadata() {
+      return this.metadata;
+    }
+
+    @JsonProperty("config")
+    public Map<String, String> config() {
+      return this.config != null ? this.config : Map.of();
+    }
+
+    @JsonProperty("storage-credentials")
+    public List<Credential> credentials() {
+      return this.credentials != null ? this.credentials : List.of();
+    }
+
+    @Override
+    public String toString() {
+      return MoreObjects.toStringHelper(this)
+          .add("metadataLocation", this.metadataLocation)
+          .add("metadata", this.metadata)
+          .add("config", this.config)
+          .toString();
+    }
+  }


Could be a record:

Suggested change

public static class StreamingLoadTableResponse implements RESTResponse {

private String metadataLocation;

private InputStream metadata;

private Map<String, String> config;

private List<Credential> credentials;

public StreamingLoadTableResponse() {}

public StreamingLoadTableResponse(

String metadataLocation,

InputStream metadata,

Map<String, String> config,

List<Credential> credentials) {

this.metadataLocation = metadataLocation;

this.metadata = metadata;

this.config = config;

this.credentials = credentials;

}

@Override

public void validate() {

Preconditions.checkNotNull(this.metadata, "Invalid metadata: null");

}

@JsonProperty("metadata-location")

public String metadataLocation() {

return this.metadataLocation;

}

@JsonProperty("metadata")

@JsonSerialize(using = InputStreamSerializer.class)

public InputStream tableMetadata() {

return this.metadata;

}

@JsonProperty("config")

public Map<String, String> config() {

return this.config != null ? this.config : Map.of();

}

@JsonProperty("storage-credentials")

public List<Credential> credentials() {

return this.credentials != null ? this.credentials : List.of();

}

@Override

public String toString() {

return MoreObjects.toStringHelper(this)

.add("metadataLocation", this.metadataLocation)

.add("metadata", this.metadata)

.add("config", this.config)

.toString();

}

}

public record StreamingLoadTableResponse(

@JsonProperty("metadata-location") String metadataLocation,

@JsonProperty("metadata") @JsonSerialize(using = InputStreamSerializer.class)

InputStream metadata,

@JsonProperty("config") Map<String, String> config,

@JsonProperty("storage-credentials") List<Credential> credentials)

implements RESTResponse {

@Override

public void validate() {

Preconditions.checkNotNull(this.metadata, "Invalid metadata: null");

}

}

adutra · 2025-04-21T13:05:52Z

...e/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

+              (fileIO, tableEntity) -> {
+                return new StreamingLoadTableResponse(
+                    tableEntity.getMetadataLocation(),
+                    fileIO.newInputFile(tableEntity.getMetadataLocation()).newStream(),
+                    Map.of(),
+                    List.of());
+              },


nit:

Suggested change

(fileIO, tableEntity) -> {

return new StreamingLoadTableResponse(

tableEntity.getMetadataLocation(),

fileIO.newInputFile(tableEntity.getMetadataLocation()).newStream(),

Map.of(),

List.of());

},

(fileIO, tableEntity) ->

new StreamingLoadTableResponse(

tableEntity.getMetadataLocation(),

fileIO.newInputFile(tableEntity.getMetadataLocation()).newStream(),

Map.of(),

List.of()),

adutra · 2025-04-21T13:18:31Z

...e/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

+    }
+  }
+
+  public static class InputStreamSerializer extends RawSerializer<InputStream> {


Nit: I would extend StdSerializer instead. In spite of its name, RawSerializer is not a generic serializer for "raw" values.

adutra · 2025-04-21T13:26:09Z

...e/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

+        throws IOException {
+      LoggerFactory.getLogger(getClass()).warn("Using custom serializer for input stream");
+      byte[] buffer = new byte[4096];
+      try (inputStream) {


Nit: I would rather model the field as an InputFile, then create a serializer for InputFile. This would make it clear when the input stream is created and closed.

Suggested change

try (inputStream) {

try (InputStream inputStream = inputFile.newStream()) {

adutra · 2025-04-21T13:44:32Z

...e/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

+        // tell jackson we're about to write a value
+        // this ensures we get the correct token prefixing the value (i.e., a : for a field value or
+        // a , for an array)
+        jsonGenerator.writeRawValue("");


The following alternative looks slightly less hacky imho:

jsonGenerator.writeRawValue(""); jsonGenerator.flush(); if (jsonGenerator.getOutputTarget() instanceof OutputStream out) { inputStream.transferTo(out); } else if (jsonGenerator.getOutputTarget() instanceof Writer writer) { new InputStreamReader(inputStream, StandardCharsets.UTF_8).transferTo(writer); } else { throw new IllegalStateException( "Cannot serialize InputFile to unknown output target: " + jsonGenerator.getOutputTarget()); }

adutra · 2025-04-21T13:48:18Z

...e/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

      TableIdentifier tableIdentifier, String snapshots) {
    return loadTableWithAccessDelegationIfStale(tableIdentifier, null, snapshots).get();
  }

+  public LoadCredentialsResponse loadAccessDelegation(


This addition looks unrelated?

adutra · 2025-04-21T13:48:39Z

...n/src/main/java/org/apache/polaris/service/catalog/iceberg/SupportsCredentialDelegation.java

@@ -36,4 +38,12 @@ Map<String, String> getCredentialConfig(
      TableIdentifier tableIdentifier,
      TableMetadata tableMetadata,
      Set<PolarisStorageActions> storageActions);
+
+  Map<String, String> getCredentialConfig(


dimas-b

Streaming metadata sounds like a valuable improvement to me 👍

dimas-b · 2025-04-21T23:37:55Z

...e/common/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

+              (fileIO, tableEntity) -> {
+                return new StreamingLoadTableResponse(
+                    tableEntity.getMetadataLocation(),
+                    fileIO.newInputFile(tableEntity.getMetadataLocation()).newStream(),


How is the stream closed? Could you add a comment about that?

eric-maynard · 2025-04-22T22:14:02Z

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -201,4 +201,11 @@ protected FeatureConfiguration(
                  + " requires experimentation in the specific deployment environment")
          .defaultValue(100 * EntityWeigher.WEIGHT_PER_MB)
          .buildFeatureConfiguration();
+
+  public static final FeatureConfiguration<Boolean> ENABLE_STREAMING_TABLE_METADATA =


Instead of just an enable/disable, can we have something like STREAMING_TABLE_METADATA_MIN_SIZE?

Initial POC - streaming table metadata rather than de/serializing in …

c420107

…memory

collado-mike requested review from adutra, ashvina, dennishuo, dimas-b, eric-maynard, jackye1995, jbonofre, vvcephei, snazy, RussellSpitzer, takidau, MonkeyCanCode, flyrain and ebyhr as code owners April 18, 2025 00:35

github-project-automation bot added this to Basic Kanban Board Apr 18, 2025

github-project-automation bot moved this to PRs In Progress in Basic Kanban Board Apr 18, 2025

adutra reviewed Apr 21, 2025

View reviewed changes

dimas-b reviewed Apr 22, 2025

View reviewed changes

eric-maynard reviewed Apr 22, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Initial POC - streaming table metadata rather than de/serializing in memory #1395

Initial POC - streaming table metadata rather than de/serializing in memory #1395

collado-mike commented Apr 18, 2025

adutra left a comment

adutra Apr 21, 2025

adutra Apr 21, 2025

adutra Apr 21, 2025

adutra Apr 21, 2025

adutra Apr 21, 2025

adutra Apr 21, 2025

adutra Apr 21, 2025

adutra Apr 21, 2025

dimas-b left a comment

dimas-b Apr 21, 2025

eric-maynard Apr 22, 2025

	try (inputStream) {
	try (InputStream inputStream = inputFile.newStream()) {

Initial POC - streaming table metadata rather than de/serializing in memory #1395

Are you sure you want to change the base?

Initial POC - streaming table metadata rather than de/serializing in memory #1395

Conversation

collado-mike commented Apr 18, 2025

adutra left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dimas-b left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment