ahmedabu98 commented on code in PR #39724:
URL: https://github.com/apache/beam/pull/39724#discussion_r3813793784
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java:
##########
@@ -59,70 +71,141 @@ public abstract class SerializableTableSpec implements
Serializable {
public abstract String getLocation();
@SchemaFieldNumber("3")
- public abstract int getSpecId();
+ public abstract int getSchemaId();
@SchemaFieldNumber("4")
- public abstract String getSchemaJson();
+ public abstract Map<Integer, String> getSchemasJson();
@SchemaFieldNumber("5")
- public abstract String getPartitionSpecJson();
+ public abstract int getSpecId();
@SchemaFieldNumber("6")
- public abstract String getSortOrderJson();
+ public abstract Map<Integer, String> getPartitionSpecsJson();
@SchemaFieldNumber("7")
+ public abstract int getOrderId();
+
+ @SchemaFieldNumber("8")
+ public abstract Map<Integer, String> getSortOrdersJson();
+
+ @SchemaFieldNumber("9")
public abstract Map<String, String> getProperties();
- private transient volatile @MonotonicNonNull Schema cachedSchema;
- private transient volatile @MonotonicNonNull PartitionSpec
cachedPartitionSpec;
- private transient volatile @MonotonicNonNull SortOrder cachedSortOrder;
+ @SchemaFieldNumber("10")
+ public abstract String getFileIoJson();
+
+ @SchemaFieldNumber("11")
+ public abstract List<String> getEncryptedKeyJsons();
+
+ private transient volatile @MonotonicNonNull Map<Integer, Schema>
cachedSchemas;
+ private transient volatile @MonotonicNonNull Map<Integer, PartitionSpec>
cachedPartitionSpecs;
+ private transient volatile @MonotonicNonNull Map<Integer, SortOrder>
cachedSortOrders;
private transient volatile @MonotonicNonNull TableIdentifier
cachedTableIdentifier;
+ private transient volatile @MonotonicNonNull FileIO cachedFileIO;
+ private transient volatile @MonotonicNonNull List<EncryptedKey>
cachedEncryptedKeys;
private static volatile @MonotonicNonNull SchemaCoder<SerializableTableSpec>
cachedCoder;
@SchemaIgnore
- public Schema getSchema() {
- Schema local = cachedSchema;
+ public Map<Integer, Schema> getSchemas() {
+ Map<Integer, Schema> local = cachedSchemas;
if (local == null) {
synchronized (this) {
- local = cachedSchema;
+ local = cachedSchemas;
if (local == null) {
- cachedSchema = local = SchemaParser.fromJson(getSchemaJson());
+ ImmutableMap.Builder<Integer, Schema> builder =
ImmutableMap.builder();
+ for (Map.Entry<Integer, String> entry : getSchemasJson().entrySet())
{
+ builder.put(entry.getKey(),
SchemaParser.fromJson(entry.getValue()));
+ }
+ cachedSchemas = local = builder.build();
}
}
}
return local;
}
Review Comment:
Same comment for the partition spec and sort order maps
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java:
##########
@@ -59,70 +71,141 @@ public abstract class SerializableTableSpec implements
Serializable {
public abstract String getLocation();
@SchemaFieldNumber("3")
- public abstract int getSpecId();
+ public abstract int getSchemaId();
@SchemaFieldNumber("4")
- public abstract String getSchemaJson();
+ public abstract Map<Integer, String> getSchemasJson();
@SchemaFieldNumber("5")
- public abstract String getPartitionSpecJson();
+ public abstract int getSpecId();
@SchemaFieldNumber("6")
- public abstract String getSortOrderJson();
+ public abstract Map<Integer, String> getPartitionSpecsJson();
@SchemaFieldNumber("7")
+ public abstract int getOrderId();
+
+ @SchemaFieldNumber("8")
+ public abstract Map<Integer, String> getSortOrdersJson();
+
+ @SchemaFieldNumber("9")
public abstract Map<String, String> getProperties();
- private transient volatile @MonotonicNonNull Schema cachedSchema;
- private transient volatile @MonotonicNonNull PartitionSpec
cachedPartitionSpec;
- private transient volatile @MonotonicNonNull SortOrder cachedSortOrder;
+ @SchemaFieldNumber("10")
+ public abstract String getFileIoJson();
+
+ @SchemaFieldNumber("11")
+ public abstract List<String> getEncryptedKeyJsons();
+
+ private transient volatile @MonotonicNonNull Map<Integer, Schema>
cachedSchemas;
+ private transient volatile @MonotonicNonNull Map<Integer, PartitionSpec>
cachedPartitionSpecs;
+ private transient volatile @MonotonicNonNull Map<Integer, SortOrder>
cachedSortOrders;
private transient volatile @MonotonicNonNull TableIdentifier
cachedTableIdentifier;
+ private transient volatile @MonotonicNonNull FileIO cachedFileIO;
+ private transient volatile @MonotonicNonNull List<EncryptedKey>
cachedEncryptedKeys;
private static volatile @MonotonicNonNull SchemaCoder<SerializableTableSpec>
cachedCoder;
@SchemaIgnore
- public Schema getSchema() {
- Schema local = cachedSchema;
+ public Map<Integer, Schema> getSchemas() {
+ Map<Integer, Schema> local = cachedSchemas;
if (local == null) {
synchronized (this) {
- local = cachedSchema;
+ local = cachedSchemas;
if (local == null) {
- cachedSchema = local = SchemaParser.fromJson(getSchemaJson());
+ ImmutableMap.Builder<Integer, Schema> builder =
ImmutableMap.builder();
+ for (Map.Entry<Integer, String> entry : getSchemasJson().entrySet())
{
+ builder.put(entry.getKey(),
SchemaParser.fromJson(entry.getValue()));
+ }
+ cachedSchemas = local = builder.build();
}
}
}
return local;
}
Review Comment:
Can we do this in a more lazy way? If the table historically has 10 schemas
and I want to get one, I should only have to serialize that one schema, not all
10.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
