From e072d6a0168ebcdbec99709131494b57b5b51825 Mon Sep 17 00:00:00 2001 From: jrmccluskey Date: Tue, 11 Aug 2026 18:03:50 +0000 Subject: [PATCH 1/2] [#39723] Implement model for Iceberg side input cache --- .../sdk/io/iceberg/SerializableTableSpec.java | 219 +++++++++++ .../beam/sdk/io/iceberg/SideInputTable.java | 346 ++++++++++++++++++ .../io/iceberg/SerializableTableSpecTest.java | 282 ++++++++++++++ .../sdk/io/iceberg/SideInputTableTest.java | 250 +++++++++++++ 4 files changed, 1097 insertions(+) create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java new file mode 100644 index 000000000000..7d2501c65381 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java @@ -0,0 +1,219 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; + +import com.google.auto.value.AutoValue; +import java.io.Serializable; +import java.util.Map; +import org.apache.beam.sdk.schemas.AutoValueSchema; +import org.apache.beam.sdk.schemas.NoSuchSchemaException; +import org.apache.beam.sdk.schemas.SchemaCoder; +import org.apache.beam.sdk.schemas.SchemaRegistry; +import org.apache.beam.sdk.schemas.annotations.DefaultSchema; +import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; +import org.apache.beam.sdk.schemas.annotations.SchemaIgnore; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.PartitionSpecParser; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.SortOrderParser; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.checkerframework.checker.nullness.qual.MonotonicNonNull; + +/** + * A serializable, lightweight representation of an Iceberg {@link Table}'s declarative metadata. + * + *

Captures the table's schema, partition spec, sort order, location, properties, and identifier. + * Suitable for broadcasting across worker nodes via Beam's side-input mechanism. + */ +@DefaultSchema(AutoValueSchema.class) +@AutoValue +public abstract class SerializableTableSpec implements Serializable { + + @SchemaFieldNumber("0") + public abstract String getTableIdentifierString(); + + @SchemaFieldNumber("1") + public abstract String getName(); + + @SchemaFieldNumber("2") + public abstract String getLocation(); + + @SchemaFieldNumber("3") + public abstract int getSpecId(); + + @SchemaFieldNumber("4") + public abstract String getSchemaJson(); + + @SchemaFieldNumber("5") + public abstract String getPartitionSpecJson(); + + @SchemaFieldNumber("6") + public abstract String getSortOrderJson(); + + @SchemaFieldNumber("7") + public abstract Map getProperties(); + + private transient volatile @MonotonicNonNull Schema cachedSchema; + private transient volatile @MonotonicNonNull PartitionSpec cachedPartitionSpec; + private transient volatile @MonotonicNonNull SortOrder cachedSortOrder; + private transient volatile @MonotonicNonNull TableIdentifier cachedTableIdentifier; + + private static volatile @MonotonicNonNull SchemaCoder cachedCoder; + + @SchemaIgnore + public Schema getSchema() { + Schema local = cachedSchema; + if (local == null) { + synchronized (this) { + local = cachedSchema; + if (local == null) { + cachedSchema = local = SchemaParser.fromJson(getSchemaJson()); + } + } + } + return local; + } + + @SchemaIgnore + public PartitionSpec getPartitionSpec() { + PartitionSpec local = cachedPartitionSpec; + if (local == null) { + synchronized (this) { + local = cachedPartitionSpec; + if (local == null) { + cachedPartitionSpec = + local = PartitionSpecParser.fromJson(getSchema(), getPartitionSpecJson()); + } + } + } + return local; + } + + @SchemaIgnore + public SortOrder getSortOrder() { + SortOrder local = cachedSortOrder; + if (local == null) { + synchronized (this) { + local = cachedSortOrder; + if (local == null) { + cachedSortOrder = local = SortOrderParser.fromJson(getSchema(), getSortOrderJson()); + } + } + } + return local; + } + + @SchemaIgnore + public TableIdentifier getTableIdentifier() { + TableIdentifier local = cachedTableIdentifier; + if (local == null) { + synchronized (this) { + local = cachedTableIdentifier; + if (local == null) { + cachedTableIdentifier = + local = IcebergUtils.parseTableIdentifier(getTableIdentifierString()); + } + } + } + return local; + } + + public static Builder builder() { + return new AutoValue_SerializableTableSpec.Builder(); + } + + public abstract Builder toBuilder(); + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setTableIdentifierString(String tableIdentifierString); + + public abstract Builder setName(String name); + + public abstract Builder setLocation(String location); + + public abstract Builder setSpecId(int specId); + + public abstract Builder setSchemaJson(String schemaJson); + + public abstract Builder setPartitionSpecJson(String partitionSpecJson); + + public abstract Builder setSortOrderJson(String sortOrderJson); + + public abstract Builder setProperties(Map properties); + + public abstract SerializableTableSpec build(); + } + + /** + * Constructs a {@link SerializableTableSpec} from a {@link Table}, using {@link Table#name()} as + * the table identifier string. + * + *

Note: When possible, prefer {@link #fromTable(TableIdentifier, Table)} to avoid catalog name + * prefix ambiguities in {@link Table#name()}. + */ + public static SerializableTableSpec fromTable(Table table) { + return fromTable(table.name(), table); + } + + /** + * Constructs a {@link SerializableTableSpec} from a {@link TableIdentifier} and a {@link Table}. + */ + public static SerializableTableSpec fromTable(TableIdentifier tableIdentifier, Table table) { + return fromTable(IcebergUtils.tableIdentifierToString(tableIdentifier), table); + } + + /** + * Constructs a {@link SerializableTableSpec} from an explicit table identifier string and a + * {@link Table}. + */ + public static SerializableTableSpec fromTable(String tableIdentifierString, Table table) { + return builder() + .setTableIdentifierString(tableIdentifierString) + .setName(table.name()) + .setLocation(table.location()) + .setSpecId(table.spec().specId()) + .setSchemaJson(SchemaParser.toJson(table.schema())) + .setPartitionSpecJson(PartitionSpecParser.toJson(table.spec())) + .setSortOrderJson(SortOrderParser.toJson(table.sortOrder())) + .setProperties(table.properties()) + .build(); + } + + /** Returns the cached {@link SchemaCoder} for {@link SerializableTableSpec}. */ + public static SchemaCoder getCoder() { + if (cachedCoder == null) { + synchronized (SerializableTableSpec.class) { + if (cachedCoder == null) { + try { + cachedCoder = + SchemaRegistry.createDefault().getSchemaCoder(SerializableTableSpec.class); + } catch (NoSuchSchemaException e) { + throw new RuntimeException(e); + } + } + } + } + return checkStateNotNull(cachedCoder); + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java new file mode 100644 index 000000000000..c318940896fa --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java @@ -0,0 +1,346 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull; + +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import org.apache.beam.sdk.annotations.Internal; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DeleteFiles; +import org.apache.iceberg.ExpireSnapshots; +import org.apache.iceberg.HistoryEntry; +import org.apache.iceberg.IncrementalAppendScan; +import org.apache.iceberg.IncrementalChangelogScan; +import org.apache.iceberg.LocationProviders; +import org.apache.iceberg.ManageSnapshots; +import org.apache.iceberg.OverwriteFiles; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.PartitionStatisticsFile; +import org.apache.iceberg.ReplacePartitions; +import org.apache.iceberg.ReplaceSortOrder; +import org.apache.iceberg.RewriteFiles; +import org.apache.iceberg.RewriteManifests; +import org.apache.iceberg.RowDelta; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.SnapshotRef; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.StatisticsFile; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableScan; +import org.apache.iceberg.Transaction; +import org.apache.iceberg.UpdateLocation; +import org.apache.iceberg.UpdatePartitionSpec; +import org.apache.iceberg.UpdateProperties; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.UpdateStatistics; +import org.apache.iceberg.encryption.EncryptionManager; +import org.apache.iceberg.encryption.PlaintextEncryptionManager; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.LocationProvider; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * A lightweight adapter that implements {@link Table} backed by a {@link SerializableTableSpec}. + * + *

Delegates declarative metadata (schema, partition specs, sort order, properties) to the + * broadcasted {@link SerializableTableSpec} and file I/O to a worker-local {@link FileIO} instance. + * + *

Mutation operations (e.g. {@code newAppend()}, {@code updateSchema()}) throw {@link + * UnsupportedOperationException} because table commits are handled centrally in {@link + * AppendFilesToTables}. + */ +@Internal +@SuppressWarnings("nullness") +public class SideInputTable implements Table { + + private final SerializableTableSpec spec; + private final FileIO fileIO; + private final EncryptionManager encryptionManager; + private final LocationProvider locationProvider; + + public SideInputTable(SerializableTableSpec spec, FileIO fileIO) { + this(spec, fileIO, PlaintextEncryptionManager.instance()); + } + + public SideInputTable( + SerializableTableSpec spec, FileIO fileIO, EncryptionManager encryptionManager) { + this.spec = checkNotNull(spec, "spec must not be null"); + this.fileIO = checkNotNull(fileIO, "fileIO must not be null"); + this.encryptionManager = checkNotNull(encryptionManager, "encryptionManager must not be null"); + this.locationProvider = + LocationProviders.locationsFor(spec.getLocation(), spec.getProperties()); + } + + public SerializableTableSpec getTableSpec() { + return spec; + } + + @Override + public String name() { + return spec.getName(); + } + + @Override + public String location() { + return spec.getLocation(); + } + + @Override + public Schema schema() { + return spec.getSchema(); + } + + @Override + public Map schemas() { + return Collections.singletonMap(spec.getSchema().schemaId(), spec.getSchema()); + } + + @Override + public PartitionSpec spec() { + return spec.getPartitionSpec(); + } + + @Override + public Map specs() { + return Collections.singletonMap(spec.getPartitionSpec().specId(), spec.getPartitionSpec()); + } + + @Override + public SortOrder sortOrder() { + return spec.getSortOrder(); + } + + @Override + public Map sortOrders() { + return Collections.singletonMap(spec.getSortOrder().orderId(), spec.getSortOrder()); + } + + @Override + public Map properties() { + return spec.getProperties(); + } + + @Override + public LocationProvider locationProvider() { + return locationProvider; + } + + @Override + public FileIO io() { + return fileIO; + } + + @Override + public EncryptionManager encryption() { + return encryptionManager; + } + + @Override + public void refresh() { + // No-op: refresh is managed by the periodic side-input update mechanism + } + + @Override + public @Nullable Snapshot currentSnapshot() { + return null; + } + + @Override + public @Nullable Snapshot snapshot(long snapshotId) { + return null; + } + + @Override + public Iterable snapshots() { + return Collections.emptyList(); + } + + @Override + public List history() { + return Collections.emptyList(); + } + + @Override + public Map refs() { + return Collections.emptyMap(); + } + + @Override + public List statisticsFiles() { + return Collections.emptyList(); + } + + @Override + public List partitionStatisticsFiles() { + return Collections.emptyList(); + } + + @Override + public TableScan newScan() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support scans directly."); + } + + @Override + public IncrementalAppendScan newIncrementalAppendScan() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support scans directly."); + } + + @Override + public IncrementalChangelogScan newIncrementalChangelogScan() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support scans directly."); + } + + @Override + public UpdateSchema updateSchema() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public UpdatePartitionSpec updateSpec() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public UpdateProperties updateProperties() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public ReplaceSortOrder replaceSortOrder() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public UpdateLocation updateLocation() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public AppendFiles newAppend() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public AppendFiles newFastAppend() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public RewriteFiles newRewrite() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public RewriteManifests rewriteManifests() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public OverwriteFiles newOverwrite() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public RowDelta newRowDelta() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public ReplacePartitions newReplacePartitions() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public DeleteFiles newDelete() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public UpdateStatistics updateStatistics() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public ExpireSnapshots expireSnapshots() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public ManageSnapshots manageSnapshots() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public Transaction newTransaction() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support table mutations."); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof SideInputTable)) { + return false; + } + SideInputTable that = (SideInputTable) o; + return Objects.equals(spec, that.spec) + && Objects.equals(fileIO, that.fileIO) + && Objects.equals(encryptionManager, that.encryptionManager); + } + + @Override + public int hashCode() { + return Objects.hash(spec, fileIO, encryptionManager); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("spec", spec) + .add("fileIO", fileIO) + .add("encryptionManager", encryptionManager) + .toString(); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java new file mode 100644 index 000000000000..8eef7ae30dd7 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java @@ -0,0 +1,282 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import org.apache.beam.sdk.schemas.SchemaCoder; +import org.apache.beam.sdk.util.CoderUtils; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.NullOrder; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SortDirection; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.types.Types; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +public class SerializableTableSpecTest { + + @Rule public TemporaryFolder tempFolder = new TemporaryFolder(); + + private Catalog catalog; + private String warehouseLocation; + + private static final Schema COMPLEX_SCHEMA = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "name", Types.StringType.get()), + optional(3, "timestamp_val", Types.TimestampType.withZone()), + optional(4, "amount", Types.DecimalType.of(10, 2)), + optional( + 5, + "nested_struct", + Types.StructType.of( + required(6, "nested_id", Types.IntegerType.get()), + optional(7, "nested_desc", Types.StringType.get()))), + optional(8, "string_list", Types.ListType.ofOptional(9, Types.StringType.get())), + optional( + 10, + "str_int_map", + Types.MapType.ofOptional(11, 12, Types.StringType.get(), Types.IntegerType.get()))); + + @Before + public void setUp() throws Exception { + warehouseLocation = "file:" + tempFolder.newFolder().getAbsolutePath(); + catalog = + CatalogUtil.loadCatalog( + CatalogUtil.ICEBERG_CATALOG_HADOOP, + "hadoop", + ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, warehouseLocation), + new Configuration()); + } + + @Test + public void testFromTableAndGettersUnpartitioned() { + TableIdentifier tableId = TableIdentifier.of("default", "unpartitioned_table"); + Table table = catalog.createTable(tableId, TestFixtures.SCHEMA); + + SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, table); + + assertEquals(IcebergUtils.tableIdentifierToString(tableId), spec.getTableIdentifierString()); + assertEquals(table.name(), spec.getName()); + assertEquals(table.location(), spec.getLocation()); + assertEquals(table.spec().specId(), spec.getSpecId()); + assertEquals(table.schema().asStruct(), spec.getSchema().asStruct()); + assertEquals(table.spec(), spec.getPartitionSpec()); + assertTrue(spec.getPartitionSpec().isUnpartitioned()); + assertEquals(table.sortOrder(), spec.getSortOrder()); + assertEquals(tableId, spec.getTableIdentifier()); + } + + @Test + public void testFromTableAndGettersPartitionedWithSortOrder() { + TableIdentifier tableId = TableIdentifier.of("default", "partitioned_table"); + PartitionSpec partitionSpec = + PartitionSpec.builderFor(COMPLEX_SCHEMA).day("timestamp_val").identity("name").build(); + SortOrder sortOrder = + SortOrder.builderFor(COMPLEX_SCHEMA) + .sortBy("id", SortDirection.ASC, NullOrder.NULLS_FIRST) + .sortBy("name", SortDirection.DESC, NullOrder.NULLS_LAST) + .build(); + Map properties = + ImmutableMap.of("write.format.default", "parquet", "custom.property", "test-val"); + + Table table = + catalog + .buildTable(tableId, COMPLEX_SCHEMA) + .withPartitionSpec(partitionSpec) + .withSortOrder(sortOrder) + .withProperties(properties) + .create(); + + SerializableTableSpec spec = SerializableTableSpec.fromTable(table); + + assertEquals(table.name(), spec.getTableIdentifierString()); + assertEquals(table.name(), spec.getName()); + assertEquals(table.location(), spec.getLocation()); + assertEquals(partitionSpec.specId(), spec.getSpecId()); + assertEquals(table.schema().asStruct(), spec.getSchema().asStruct()); + assertEquals(partitionSpec, spec.getPartitionSpec()); + assertEquals(sortOrder, spec.getSortOrder()); + assertEquals( + properties.get("write.format.default"), spec.getProperties().get("write.format.default")); + assertEquals(properties.get("custom.property"), spec.getProperties().get("custom.property")); + } + + @Test + public void testDottedNestedNamespaceIdentifier() { + TableIdentifier tableId = TableIdentifier.of("my", "nested", "catalog", "deep_table"); + Table table = catalog.createTable(tableId, TestFixtures.SCHEMA); + + SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, table); + + assertEquals("my.nested.catalog.deep_table", spec.getTableIdentifierString()); + assertEquals(tableId, spec.getTableIdentifier()); + } + + @Test + public void testBuilderAndToBuilderWithEmptyProperties() { + TableIdentifier tableId = TableIdentifier.of("default", "empty_prop_table"); + Table table = catalog.createTable(tableId, TestFixtures.SCHEMA); + + SerializableTableSpec spec = + SerializableTableSpec.fromTable(tableId, table) + .toBuilder() + .setProperties(Collections.emptyMap()) + .build(); + + assertTrue(spec.getProperties().isEmpty()); + assertEquals(tableId, spec.getTableIdentifier()); + } + + @Test + public void testJavaSerializationRoundtrip() throws Exception { + TableIdentifier tableId = TableIdentifier.of("default", "ser_table"); + PartitionSpec partitionSpec = + PartitionSpec.builderFor(COMPLEX_SCHEMA).bucket("name", 16).build(); + Table table = + catalog.buildTable(tableId, COMPLEX_SCHEMA).withPartitionSpec(partitionSpec).create(); + + SerializableTableSpec original = SerializableTableSpec.fromTable(tableId, table); + + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + try (ObjectOutputStream oos = new ObjectOutputStream(baos)) { + oos.writeObject(original); + } + + SerializableTableSpec deserialized; + try (ObjectInputStream ois = + new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray()))) { + deserialized = (SerializableTableSpec) ois.readObject(); + } + + assertNotNull(deserialized); + assertEquals(original, deserialized); + assertEquals(original.getTableIdentifierString(), deserialized.getTableIdentifierString()); + assertEquals(original.getSchema().asStruct(), deserialized.getSchema().asStruct()); + assertEquals(original.getPartitionSpec(), deserialized.getPartitionSpec()); + assertEquals(original.getSortOrder(), deserialized.getSortOrder()); + } + + @Test + public void testBeamSchemaCoderRoundtrip() throws Exception { + TableIdentifier tableId = TableIdentifier.of("default", "coder_table"); + PartitionSpec partitionSpec = + PartitionSpec.builderFor(COMPLEX_SCHEMA).hour("timestamp_val").build(); + SortOrder sortOrder = + SortOrder.builderFor(COMPLEX_SCHEMA) + .sortBy("id", SortDirection.DESC, NullOrder.NULLS_LAST) + .build(); + Table table = + catalog + .buildTable(tableId, COMPLEX_SCHEMA) + .withPartitionSpec(partitionSpec) + .withSortOrder(sortOrder) + .withProperties(ImmutableMap.of("k1", "v1")) + .create(); + + SerializableTableSpec original = SerializableTableSpec.fromTable(tableId, table); + SchemaCoder coder = SerializableTableSpec.getCoder(); + + SerializableTableSpec decoded = CoderUtils.clone(coder, original); + + assertNotNull(decoded); + assertEquals(original, decoded); + assertEquals(original.getTableIdentifierString(), decoded.getTableIdentifierString()); + assertEquals(original.getName(), decoded.getName()); + assertEquals(original.getLocation(), decoded.getLocation()); + assertEquals(original.getSpecId(), decoded.getSpecId()); + assertEquals(original.getSchema().asStruct(), decoded.getSchema().asStruct()); + assertEquals(original.getPartitionSpec(), decoded.getPartitionSpec()); + assertEquals(original.getSortOrder(), decoded.getSortOrder()); + assertEquals(original.getProperties(), decoded.getProperties()); + } + + @Test + @SuppressWarnings("ReferenceEquality") + public void testConcurrentGetterInitializationThreadSafety() throws Exception { + TableIdentifier tableId = TableIdentifier.of("default", "concurrent_table"); + PartitionSpec partitionSpec = + PartitionSpec.builderFor(COMPLEX_SCHEMA).bucket("name", 8).build(); + Table table = + catalog.buildTable(tableId, COMPLEX_SCHEMA).withPartitionSpec(partitionSpec).create(); + + SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, table); + + int numThreads = 16; + ExecutorService executor = Executors.newFixedThreadPool(numThreads); + CountDownLatch startLatch = new CountDownLatch(1); + List> futures = new ArrayList<>(); + + try { + for (int i = 0; i < numThreads; i++) { + futures.add( + executor.submit( + () -> { + startLatch.await(); + Schema schema = spec.getSchema(); + PartitionSpec ps = spec.getPartitionSpec(); + SortOrder so = spec.getSortOrder(); + TableIdentifier ti = spec.getTableIdentifier(); + + if (schema == null || ps == null || so == null || ti == null) { + throw new IllegalStateException("Getter returned null"); + } + if (schema != spec.getSchema() || ps != spec.getPartitionSpec()) { + throw new IllegalStateException("Getter returned non-identical instance"); + } + return null; + })); + } + + startLatch.countDown(); + for (Future future : futures) { + future.get(10, TimeUnit.SECONDS); + } + } finally { + executor.shutdown(); + } + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java new file mode 100644 index 000000000000..70da317051c6 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java @@ -0,0 +1,250 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.util.Collections; +import java.util.Map; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.NullOrder; +import org.apache.iceberg.PartitionKey; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.SortDirection; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.encryption.PlaintextEncryptionManager; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +public class SideInputTableTest { + + @Rule public TemporaryFolder tempFolder = new TemporaryFolder(); + + private Catalog catalog; + private String warehouseLocation; + + @Before + public void setUp() throws Exception { + warehouseLocation = "file:" + tempFolder.newFolder().getAbsolutePath(); + catalog = + CatalogUtil.loadCatalog( + CatalogUtil.ICEBERG_CATALOG_HADOOP, + "hadoop", + ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, warehouseLocation), + new Configuration()); + } + + @Test + public void testConstructorNullChecks() { + TableIdentifier tableId = TableIdentifier.of("default", "null_check_table"); + Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA); + SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); + + assertThrows(NullPointerException.class, () -> new SideInputTable(null, realTable.io())); + assertThrows(NullPointerException.class, () -> new SideInputTable(spec, null)); + assertThrows(NullPointerException.class, () -> new SideInputTable(spec, realTable.io(), null)); + } + + @Test + public void testMetadataDelegation() { + TableIdentifier tableId = TableIdentifier.of("default", "side_input_test_table"); + PartitionSpec partitionSpec = + PartitionSpec.builderFor(TestFixtures.SCHEMA).identity("data").build(); + SortOrder sortOrder = + SortOrder.builderFor(TestFixtures.SCHEMA) + .sortBy("id", SortDirection.ASC, NullOrder.NULLS_FIRST) + .build(); + Map properties = + ImmutableMap.of("write.format.default", "parquet", "user.key", "user.val"); + + Table realTable = + catalog + .buildTable(tableId, TestFixtures.SCHEMA) + .withPartitionSpec(partitionSpec) + .withSortOrder(sortOrder) + .withProperties(properties) + .create(); + + SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); + SideInputTable sideInputTable = + new SideInputTable(spec, realTable.io(), PlaintextEncryptionManager.instance()); + + assertEquals(realTable.name(), sideInputTable.name()); + assertEquals(realTable.location(), sideInputTable.location()); + assertEquals(realTable.schema().asStruct(), sideInputTable.schema().asStruct()); + assertEquals(realTable.schemas().keySet(), sideInputTable.schemas().keySet()); + assertEquals(realTable.spec(), sideInputTable.spec()); + assertEquals(realTable.specs().keySet(), sideInputTable.specs().keySet()); + assertEquals(realTable.sortOrder(), sideInputTable.sortOrder()); + assertEquals(realTable.sortOrders().keySet(), sideInputTable.sortOrders().keySet()); + assertEquals( + realTable.properties().get("user.key"), sideInputTable.properties().get("user.key")); + assertEquals(realTable.io(), sideInputTable.io()); + assertNotNull(sideInputTable.locationProvider()); + assertNotNull(sideInputTable.encryption()); + assertEquals(spec, sideInputTable.getTableSpec()); + assertTrue(sideInputTable.specs().containsKey(spec.getPartitionSpec().specId())); + + // Verify refresh is a safe no-op + sideInputTable.refresh(); + } + + @Test + public void testSnapshotQueriesReturnEmptyOrNull() { + TableIdentifier tableId = TableIdentifier.of("default", "snapshot_query_table"); + Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA); + SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); + SideInputTable sideInputTable = new SideInputTable(spec, realTable.io()); + + assertNull(sideInputTable.currentSnapshot()); + assertNull(sideInputTable.snapshot(12345L)); + assertEquals(Collections.emptyList(), ImmutableList.copyOf(sideInputTable.snapshots())); + assertEquals(Collections.emptyList(), sideInputTable.history()); + assertEquals(Collections.emptyMap(), sideInputTable.refs()); + assertEquals(Collections.emptyList(), sideInputTable.statisticsFiles()); + assertEquals(Collections.emptyList(), sideInputTable.partitionStatisticsFiles()); + } + + @Test + public void testWritingPartitionedWithRecordWriter() throws Exception { + TableIdentifier tableId = TableIdentifier.of("default", "partitioned_writer_table"); + PartitionSpec partitionSpec = + PartitionSpec.builderFor(TestFixtures.SCHEMA).identity("data").build(); + Table realTable = + catalog.buildTable(tableId, TestFixtures.SCHEMA).withPartitionSpec(partitionSpec).create(); + + SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); + SideInputTable sideInputTable = new SideInputTable(spec, realTable.io()); + + PartitionKey partitionKey = new PartitionKey(sideInputTable.spec(), sideInputTable.schema()); + Record record = GenericRecord.create(sideInputTable.schema()); + record.setField("id", 42L); + record.setField("data", "test_partition_value"); + partitionKey.partition(record); + + RecordWriter writer = + new RecordWriter( + sideInputTable, FileFormat.PARQUET, "test_file_001", partitionKey, ImmutableMap.of()); + + writer.write(record); + writer.close(); + + assertNotNull(writer.getDataFile()); + assertNotNull(writer.getDataFile().path()); + assertEquals(1, writer.getDataFile().recordCount()); + assertEquals(FileFormat.PARQUET, writer.getDataFile().format()); + } + + @Test + public void testWritingUnpartitionedWithRecordWriter() throws Exception { + TableIdentifier tableId = TableIdentifier.of("default", "unpartitioned_writer_table"); + Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA); + + SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); + SideInputTable sideInputTable = new SideInputTable(spec, realTable.io()); + + PartitionKey partitionKey = new PartitionKey(sideInputTable.spec(), sideInputTable.schema()); + Record record = GenericRecord.create(sideInputTable.schema()); + record.setField("id", 99L); + record.setField("data", "unpartitioned_data"); + partitionKey.partition(record); + + RecordWriter writer = + new RecordWriter( + sideInputTable, + FileFormat.PARQUET, + "test_unpartitioned_file_001", + partitionKey, + ImmutableMap.of()); + + writer.write(record); + writer.close(); + + assertNotNull(writer.getDataFile()); + assertNotNull(writer.getDataFile().path()); + assertEquals(1, writer.getDataFile().recordCount()); + assertEquals(FileFormat.PARQUET, writer.getDataFile().format()); + } + + @Test + public void testUnsupportedOperationsThrowExceptions() { + TableIdentifier tableId = TableIdentifier.of("default", "mutations_test_table"); + Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA); + SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); + SideInputTable sideInputTable = new SideInputTable(spec, realTable.io()); + + assertThrows(UnsupportedOperationException.class, sideInputTable::newScan); + assertThrows(UnsupportedOperationException.class, sideInputTable::newIncrementalAppendScan); + assertThrows(UnsupportedOperationException.class, sideInputTable::newIncrementalChangelogScan); + assertThrows(UnsupportedOperationException.class, sideInputTable::updateSchema); + assertThrows(UnsupportedOperationException.class, sideInputTable::updateSpec); + assertThrows(UnsupportedOperationException.class, sideInputTable::updateProperties); + assertThrows(UnsupportedOperationException.class, sideInputTable::replaceSortOrder); + assertThrows(UnsupportedOperationException.class, sideInputTable::updateLocation); + assertThrows(UnsupportedOperationException.class, sideInputTable::newAppend); + assertThrows(UnsupportedOperationException.class, sideInputTable::newFastAppend); + assertThrows(UnsupportedOperationException.class, sideInputTable::newRewrite); + assertThrows(UnsupportedOperationException.class, sideInputTable::rewriteManifests); + assertThrows(UnsupportedOperationException.class, sideInputTable::newOverwrite); + assertThrows(UnsupportedOperationException.class, sideInputTable::newRowDelta); + assertThrows(UnsupportedOperationException.class, sideInputTable::newReplacePartitions); + assertThrows(UnsupportedOperationException.class, sideInputTable::newDelete); + assertThrows(UnsupportedOperationException.class, sideInputTable::updateStatistics); + assertThrows(UnsupportedOperationException.class, sideInputTable::expireSnapshots); + assertThrows(UnsupportedOperationException.class, sideInputTable::manageSnapshots); + assertThrows(UnsupportedOperationException.class, sideInputTable::newTransaction); + } + + @Test + public void testEqualsHashCodeAndToString() { + TableIdentifier tableId1 = TableIdentifier.of("default", "t1"); + TableIdentifier tableId2 = TableIdentifier.of("default", "t2"); + Table realTable1 = catalog.createTable(tableId1, TestFixtures.SCHEMA); + Table realTable2 = catalog.createTable(tableId2, TestFixtures.SCHEMA); + + SerializableTableSpec spec1 = SerializableTableSpec.fromTable(tableId1, realTable1); + SerializableTableSpec spec2 = SerializableTableSpec.fromTable(tableId2, realTable2); + + SideInputTable table1a = new SideInputTable(spec1, realTable1.io()); + SideInputTable table1b = new SideInputTable(spec1, realTable1.io()); + SideInputTable table2 = new SideInputTable(spec2, realTable2.io()); + + assertEquals(table1a, table1b); + assertEquals(table1a.hashCode(), table1b.hashCode()); + assertNotEquals(table1a, table2); + assertNotNull(table1a.toString()); + assertTrue(table1a.toString().contains("SideInputTable")); + } +} From aec49bab158a03e62191cac9aadc562fa14c839c Mon Sep 17 00:00:00 2001 From: jrmccluskey Date: Tue, 18 Aug 2026 19:13:53 +0000 Subject: [PATCH 2/2] Move FileIO and EncryptionManager, address comments --- .../sdk/io/iceberg/SerializableTableSpec.java | 219 +++++++++++++++--- .../beam/sdk/io/iceberg/SideInputTable.java | 80 ++++--- .../io/iceberg/SerializableTableSpecTest.java | 72 +++++- .../sdk/io/iceberg/SideInputTableTest.java | 65 +++--- 4 files changed, 338 insertions(+), 98 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java index 7d2501c65381..c6ee4a976993 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java @@ -21,7 +21,10 @@ import com.google.auto.value.AutoValue; import java.io.Serializable; +import java.util.Collections; +import java.util.List; import java.util.Map; +import java.util.stream.Collectors; import org.apache.beam.sdk.schemas.AutoValueSchema; import org.apache.beam.sdk.schemas.NoSuchSchemaException; import org.apache.beam.sdk.schemas.SchemaCoder; @@ -29,6 +32,9 @@ import org.apache.beam.sdk.schemas.annotations.DefaultSchema; import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; import org.apache.beam.sdk.schemas.annotations.SchemaIgnore; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.EncryptedKeyParser; +import org.apache.iceberg.HasTableOperations; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.PartitionSpecParser; import org.apache.iceberg.Schema; @@ -36,14 +42,20 @@ import org.apache.iceberg.SortOrder; import org.apache.iceberg.SortOrderParser; import org.apache.iceberg.Table; +import org.apache.iceberg.TableMetadata; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.encryption.EncryptedKey; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.FileIOParser; import org.checkerframework.checker.nullness.qual.MonotonicNonNull; +import org.checkerframework.checker.nullness.qual.Nullable; /** * A serializable, lightweight representation of an Iceberg {@link Table}'s declarative metadata. * - *

Captures the table's schema, partition spec, sort order, location, properties, and identifier. - * Suitable for broadcasting across worker nodes via Beam's side-input mechanism. + *

Captures the table's schemas, partition specs, sort orders, location, properties, identifier, + * encrypted keys, and serialized {@link FileIO} configuration. Suitable for broadcasting across + * worker nodes via Beam's side-input mechanism. */ @DefaultSchema(AutoValueSchema.class) @AutoValue @@ -59,35 +71,53 @@ 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 getSchemasJson(); @SchemaFieldNumber("5") - public abstract String getPartitionSpecJson(); + public abstract int getSpecId(); @SchemaFieldNumber("6") - public abstract String getSortOrderJson(); + public abstract Map getPartitionSpecsJson(); @SchemaFieldNumber("7") + public abstract int getOrderId(); + + @SchemaFieldNumber("8") + public abstract Map getSortOrdersJson(); + + @SchemaFieldNumber("9") public abstract Map 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 getEncryptedKeyJsons(); + + private transient volatile @MonotonicNonNull Map cachedSchemas; + private transient volatile @MonotonicNonNull Map cachedPartitionSpecs; + private transient volatile @MonotonicNonNull Map cachedSortOrders; private transient volatile @MonotonicNonNull TableIdentifier cachedTableIdentifier; + private transient volatile @MonotonicNonNull FileIO cachedFileIO; + private transient volatile @MonotonicNonNull List cachedEncryptedKeys; private static volatile @MonotonicNonNull SchemaCoder cachedCoder; @SchemaIgnore - public Schema getSchema() { - Schema local = cachedSchema; + public Map getSchemas() { + Map local = cachedSchemas; if (local == null) { synchronized (this) { - local = cachedSchema; + local = cachedSchemas; if (local == null) { - cachedSchema = local = SchemaParser.fromJson(getSchemaJson()); + ImmutableMap.Builder builder = ImmutableMap.builder(); + for (Map.Entry entry : getSchemasJson().entrySet()) { + builder.put(entry.getKey(), SchemaParser.fromJson(entry.getValue())); + } + cachedSchemas = local = builder.build(); } } } @@ -95,14 +125,33 @@ public Schema getSchema() { } @SchemaIgnore - public PartitionSpec getPartitionSpec() { - PartitionSpec local = cachedPartitionSpec; + public Schema getSchema() { + Schema schema = getSchemas().get(getSchemaId()); + if (schema == null) { + throw new IllegalStateException( + "Schema with id " + getSchemaId() + " not found in schemas map"); + } + return schema; + } + + @SchemaIgnore + public @Nullable Schema getSchema(int schemaId) { + return getSchemas().get(schemaId); + } + + @SchemaIgnore + public Map getPartitionSpecs() { + Map local = cachedPartitionSpecs; if (local == null) { synchronized (this) { - local = cachedPartitionSpec; + local = cachedPartitionSpecs; if (local == null) { - cachedPartitionSpec = - local = PartitionSpecParser.fromJson(getSchema(), getPartitionSpecJson()); + ImmutableMap.Builder builder = ImmutableMap.builder(); + for (Map.Entry entry : getPartitionSpecsJson().entrySet()) { + builder.put( + entry.getKey(), PartitionSpecParser.fromJson(getSchema(), entry.getValue())); + } + cachedPartitionSpecs = local = builder.build(); } } } @@ -110,19 +159,53 @@ public PartitionSpec getPartitionSpec() { } @SchemaIgnore - public SortOrder getSortOrder() { - SortOrder local = cachedSortOrder; + public PartitionSpec getPartitionSpec() { + PartitionSpec spec = getPartitionSpecs().get(getSpecId()); + if (spec == null) { + throw new IllegalStateException( + "PartitionSpec with id " + getSpecId() + " not found in partitionSpecs map"); + } + return spec; + } + + @SchemaIgnore + public @Nullable PartitionSpec getPartitionSpec(int specId) { + return getPartitionSpecs().get(specId); + } + + @SchemaIgnore + public Map getSortOrders() { + Map local = cachedSortOrders; if (local == null) { synchronized (this) { - local = cachedSortOrder; + local = cachedSortOrders; if (local == null) { - cachedSortOrder = local = SortOrderParser.fromJson(getSchema(), getSortOrderJson()); + ImmutableMap.Builder builder = ImmutableMap.builder(); + for (Map.Entry entry : getSortOrdersJson().entrySet()) { + builder.put(entry.getKey(), SortOrderParser.fromJson(getSchema(), entry.getValue())); + } + cachedSortOrders = local = builder.build(); } } } return local; } + @SchemaIgnore + public SortOrder getSortOrder() { + SortOrder order = getSortOrders().get(getOrderId()); + if (order == null) { + throw new IllegalStateException( + "SortOrder with id " + getOrderId() + " not found in sortOrders map"); + } + return order; + } + + @SchemaIgnore + public @Nullable SortOrder getSortOrder(int orderId) { + return getSortOrders().get(orderId); + } + @SchemaIgnore public TableIdentifier getTableIdentifier() { TableIdentifier local = cachedTableIdentifier; @@ -138,6 +221,38 @@ public TableIdentifier getTableIdentifier() { return local; } + @SchemaIgnore + public FileIO getFileIO() { + FileIO local = cachedFileIO; + if (local == null) { + synchronized (this) { + local = cachedFileIO; + if (local == null) { + cachedFileIO = local = FileIOParser.fromJson(getFileIoJson()); + } + } + } + return local; + } + + @SchemaIgnore + public List getEncryptedKeys() { + List local = cachedEncryptedKeys; + if (local == null) { + synchronized (this) { + local = cachedEncryptedKeys; + if (local == null) { + cachedEncryptedKeys = + local = + getEncryptedKeyJsons().stream() + .map(EncryptedKeyParser::fromJson) + .collect(Collectors.toList()); + } + } + } + return local; + } + public static Builder builder() { return new AutoValue_SerializableTableSpec.Builder(); } @@ -152,16 +267,29 @@ public abstract static class Builder { public abstract Builder setLocation(String location); + public abstract Builder setSchemaId(int schemaId); + + public abstract Builder setSchemasJson(Map schemasJson); + public abstract Builder setSpecId(int specId); - public abstract Builder setSchemaJson(String schemaJson); + public abstract Builder setPartitionSpecsJson(Map partitionSpecsJson); - public abstract Builder setPartitionSpecJson(String partitionSpecJson); + public abstract Builder setOrderId(int orderId); - public abstract Builder setSortOrderJson(String sortOrderJson); + public abstract Builder setSortOrdersJson(Map sortOrdersJson); public abstract Builder setProperties(Map properties); + public abstract Builder setFileIoJson(String fileIoJson); + + public abstract Builder setEncryptedKeyJsons(List encryptedKeyJsons); + + @SchemaIgnore + public Builder setFileIO(FileIO fileIO) { + return setFileIoJson(FileIOParser.toJson(fileIO)); + } + public abstract SerializableTableSpec build(); } @@ -188,15 +316,50 @@ public static SerializableTableSpec fromTable(TableIdentifier tableIdentifier, T * {@link Table}. */ public static SerializableTableSpec fromTable(String tableIdentifierString, Table table) { + if (!(table instanceof HasTableOperations)) { + throw new IllegalArgumentException( + String.format( + "Table %s of class %s does not implement HasTableOperations", + table.name(), table.getClass().getName())); + } + + TableMetadata metadata = ((HasTableOperations) table).operations().current(); + List encryptedKeyJsons = Collections.emptyList(); + if (metadata != null && metadata.encryptionKeys() != null) { + encryptedKeyJsons = + metadata.encryptionKeys().stream() + .map(key -> EncryptedKeyParser.toJson(key, false)) + .collect(Collectors.toList()); + } + + ImmutableMap.Builder schemasJson = ImmutableMap.builder(); + for (Map.Entry entry : table.schemas().entrySet()) { + schemasJson.put(entry.getKey(), SchemaParser.toJson(entry.getValue())); + } + + ImmutableMap.Builder specsJson = ImmutableMap.builder(); + for (Map.Entry entry : table.specs().entrySet()) { + specsJson.put(entry.getKey(), PartitionSpecParser.toJson(entry.getValue())); + } + + ImmutableMap.Builder sortOrdersJson = ImmutableMap.builder(); + for (Map.Entry entry : table.sortOrders().entrySet()) { + sortOrdersJson.put(entry.getKey(), SortOrderParser.toJson(entry.getValue())); + } + return builder() .setTableIdentifierString(tableIdentifierString) .setName(table.name()) .setLocation(table.location()) + .setSchemaId(table.schema().schemaId()) + .setSchemasJson(schemasJson.build()) .setSpecId(table.spec().specId()) - .setSchemaJson(SchemaParser.toJson(table.schema())) - .setPartitionSpecJson(PartitionSpecParser.toJson(table.spec())) - .setSortOrderJson(SortOrderParser.toJson(table.sortOrder())) + .setPartitionSpecsJson(specsJson.build()) + .setOrderId(table.sortOrder().orderId()) + .setSortOrdersJson(sortOrdersJson.build()) .setProperties(table.properties()) + .setFileIoJson(FileIOParser.toJson(table.io())) + .setEncryptedKeyJsons(encryptedKeyJsons) .build(); } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java index c318940896fa..aa51571c4cbd 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java @@ -47,6 +47,7 @@ import org.apache.iceberg.SortOrder; import org.apache.iceberg.StatisticsFile; import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; import org.apache.iceberg.TableScan; import org.apache.iceberg.Transaction; import org.apache.iceberg.UpdateLocation; @@ -55,38 +56,53 @@ import org.apache.iceberg.UpdateSchema; import org.apache.iceberg.UpdateStatistics; import org.apache.iceberg.encryption.EncryptionManager; +import org.apache.iceberg.encryption.EncryptionUtil; +import org.apache.iceberg.encryption.KeyManagementClient; import org.apache.iceberg.encryption.PlaintextEncryptionManager; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.LocationProvider; -import org.checkerframework.checker.nullness.qual.Nullable; /** * A lightweight adapter that implements {@link Table} backed by a {@link SerializableTableSpec}. * - *

Delegates declarative metadata (schema, partition specs, sort order, properties) to the - * broadcasted {@link SerializableTableSpec} and file I/O to a worker-local {@link FileIO} instance. + *

Delegates declarative metadata (schemas, partition specs, sort orders, properties) and {@link + * FileIO} to the broadcasted {@link SerializableTableSpec}, and reconstructs the {@link + * EncryptionManager} from catalog properties or falls back to {@link PlaintextEncryptionManager}. * - *

Mutation operations (e.g. {@code newAppend()}, {@code updateSchema()}) throw {@link - * UnsupportedOperationException} because table commits are handled centrally in {@link - * AppendFilesToTables}. + *

All non-metadata or mutating operations (e.g. {@code refresh()}, {@code currentSnapshot()}, + * {@code newAppend()}, {@code updateSchema()}) throw {@link UnsupportedOperationException}. Table + * commits are handled centrally in {@link AppendFilesToTables}. */ @Internal @SuppressWarnings("nullness") public class SideInputTable implements Table { private final SerializableTableSpec spec; - private final FileIO fileIO; private final EncryptionManager encryptionManager; private final LocationProvider locationProvider; - public SideInputTable(SerializableTableSpec spec, FileIO fileIO) { - this(spec, fileIO, PlaintextEncryptionManager.instance()); + public SideInputTable(SerializableTableSpec spec) { + this(spec, Collections.emptyMap()); } - public SideInputTable( - SerializableTableSpec spec, FileIO fileIO, EncryptionManager encryptionManager) { + public SideInputTable(SerializableTableSpec spec, Map catalogProperties) { + this.spec = checkNotNull(spec, "spec must not be null"); + checkNotNull(catalogProperties, "catalogProperties must not be null"); + this.locationProvider = + LocationProviders.locationsFor(spec.getLocation(), spec.getProperties()); + + Map properties = spec.getProperties(); + if (!properties.containsKey(TableProperties.ENCRYPTION_TABLE_KEY)) { + this.encryptionManager = PlaintextEncryptionManager.instance(); + } else { + KeyManagementClient kmsClient = EncryptionUtil.createKmsClient(catalogProperties); + this.encryptionManager = + EncryptionUtil.createEncryptionManager(spec.getEncryptedKeys(), properties, kmsClient); + } + } + + public SideInputTable(SerializableTableSpec spec, EncryptionManager encryptionManager) { this.spec = checkNotNull(spec, "spec must not be null"); - this.fileIO = checkNotNull(fileIO, "fileIO must not be null"); this.encryptionManager = checkNotNull(encryptionManager, "encryptionManager must not be null"); this.locationProvider = LocationProviders.locationsFor(spec.getLocation(), spec.getProperties()); @@ -113,7 +129,7 @@ public Schema schema() { @Override public Map schemas() { - return Collections.singletonMap(spec.getSchema().schemaId(), spec.getSchema()); + return spec.getSchemas(); } @Override @@ -123,7 +139,7 @@ public PartitionSpec spec() { @Override public Map specs() { - return Collections.singletonMap(spec.getPartitionSpec().specId(), spec.getPartitionSpec()); + return spec.getPartitionSpecs(); } @Override @@ -133,7 +149,7 @@ public SortOrder sortOrder() { @Override public Map sortOrders() { - return Collections.singletonMap(spec.getSortOrder().orderId(), spec.getSortOrder()); + return spec.getSortOrders(); } @Override @@ -148,7 +164,7 @@ public LocationProvider locationProvider() { @Override public FileIO io() { - return fileIO; + return spec.getFileIO(); } @Override @@ -158,42 +174,50 @@ public EncryptionManager encryption() { @Override public void refresh() { - // No-op: refresh is managed by the periodic side-input update mechanism + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support refresh."); } @Override - public @Nullable Snapshot currentSnapshot() { - return null; + public Snapshot currentSnapshot() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support snapshots."); } @Override - public @Nullable Snapshot snapshot(long snapshotId) { - return null; + public Snapshot snapshot(long snapshotId) { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support snapshots."); } @Override public Iterable snapshots() { - return Collections.emptyList(); + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support snapshots."); } @Override public List history() { - return Collections.emptyList(); + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support snapshots."); } @Override public Map refs() { - return Collections.emptyMap(); + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support snapshot refs."); } @Override public List statisticsFiles() { - return Collections.emptyList(); + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support statisticsFiles."); } @Override public List partitionStatisticsFiles() { - return Collections.emptyList(); + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support partitionStatisticsFiles."); } @Override @@ -326,20 +350,18 @@ public boolean equals(Object o) { } SideInputTable that = (SideInputTable) o; return Objects.equals(spec, that.spec) - && Objects.equals(fileIO, that.fileIO) && Objects.equals(encryptionManager, that.encryptionManager); } @Override public int hashCode() { - return Objects.hash(spec, fileIO, encryptionManager); + return Objects.hash(spec, encryptionManager); } @Override public String toString() { return MoreObjects.toStringHelper(this) .add("spec", spec) - .add("fileIO", fileIO) .add("encryptionManager", encryptionManager) .toString(); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java index 8eef7ae30dd7..87a843db7a2f 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java @@ -21,7 +21,10 @@ import static org.apache.iceberg.types.Types.NestedField.required; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; @@ -50,6 +53,8 @@ import org.apache.iceberg.Table; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.encryption.EncryptedKey; +import org.apache.iceberg.io.FileIO; import org.apache.iceberg.types.Types; import org.junit.Before; import org.junit.Rule; @@ -92,6 +97,15 @@ public void setUp() throws Exception { new Configuration()); } + @Test + public void testFromTableThrowsWhenNotImplementingHasTableOperations() { + Table mockTable = mock(Table.class); + when(mockTable.name()).thenReturn("mock_table"); + assertThrows( + IllegalArgumentException.class, + () -> SerializableTableSpec.fromTable(TableIdentifier.of("default", "mock"), mockTable)); + } + @Test public void testFromTableAndGettersUnpartitioned() { TableIdentifier tableId = TableIdentifier.of("default", "unpartitioned_table"); @@ -102,12 +116,22 @@ public void testFromTableAndGettersUnpartitioned() { assertEquals(IcebergUtils.tableIdentifierToString(tableId), spec.getTableIdentifierString()); assertEquals(table.name(), spec.getName()); assertEquals(table.location(), spec.getLocation()); + assertEquals(table.schema().schemaId(), spec.getSchemaId()); assertEquals(table.spec().specId(), spec.getSpecId()); + assertEquals(table.sortOrder().orderId(), spec.getOrderId()); assertEquals(table.schema().asStruct(), spec.getSchema().asStruct()); + assertEquals(table.schemas().size(), spec.getSchemas().size()); assertEquals(table.spec(), spec.getPartitionSpec()); + assertEquals(table.specs().size(), spec.getPartitionSpecs().size()); assertTrue(spec.getPartitionSpec().isUnpartitioned()); assertEquals(table.sortOrder(), spec.getSortOrder()); + assertEquals(table.sortOrders().size(), spec.getSortOrders().size()); assertEquals(tableId, spec.getTableIdentifier()); + assertNotNull(spec.getFileIO()); + assertEquals(table.io().getClass().getName(), spec.getFileIO().getClass().getName()); + assertNotNull(spec.getEncryptedKeyJsons()); + assertNotNull(spec.getEncryptedKeys()); + assertTrue(spec.getEncryptedKeys().isEmpty()); } @Test @@ -136,13 +160,20 @@ public void testFromTableAndGettersPartitionedWithSortOrder() { assertEquals(table.name(), spec.getTableIdentifierString()); assertEquals(table.name(), spec.getName()); assertEquals(table.location(), spec.getLocation()); + assertEquals(table.schema().schemaId(), spec.getSchemaId()); assertEquals(partitionSpec.specId(), spec.getSpecId()); + assertEquals(sortOrder.orderId(), spec.getOrderId()); assertEquals(table.schema().asStruct(), spec.getSchema().asStruct()); + assertEquals(table.schemas().keySet(), spec.getSchemas().keySet()); assertEquals(partitionSpec, spec.getPartitionSpec()); + assertEquals(table.specs().keySet(), spec.getPartitionSpecs().keySet()); assertEquals(sortOrder, spec.getSortOrder()); + assertEquals(table.sortOrders().keySet(), spec.getSortOrders().keySet()); assertEquals( properties.get("write.format.default"), spec.getProperties().get("write.format.default")); assertEquals(properties.get("custom.property"), spec.getProperties().get("custom.property")); + assertNotNull(spec.getFileIO()); + assertNotNull(spec.getEncryptedKeys()); } @Test @@ -169,6 +200,7 @@ public void testBuilderAndToBuilderWithEmptyProperties() { assertTrue(spec.getProperties().isEmpty()); assertEquals(tableId, spec.getTableIdentifier()); + assertNotNull(spec.getFileIO()); } @Test @@ -195,9 +227,18 @@ public void testJavaSerializationRoundtrip() throws Exception { assertNotNull(deserialized); assertEquals(original, deserialized); assertEquals(original.getTableIdentifierString(), deserialized.getTableIdentifierString()); + assertEquals(original.getSchemaId(), deserialized.getSchemaId()); + assertEquals(original.getSpecId(), deserialized.getSpecId()); + assertEquals(original.getOrderId(), deserialized.getOrderId()); assertEquals(original.getSchema().asStruct(), deserialized.getSchema().asStruct()); + assertEquals(original.getSchemas().keySet(), deserialized.getSchemas().keySet()); assertEquals(original.getPartitionSpec(), deserialized.getPartitionSpec()); + assertEquals(original.getPartitionSpecs().keySet(), deserialized.getPartitionSpecs().keySet()); assertEquals(original.getSortOrder(), deserialized.getSortOrder()); + assertEquals(original.getSortOrders().keySet(), deserialized.getSortOrders().keySet()); + assertEquals(original.getEncryptedKeyJsons(), deserialized.getEncryptedKeyJsons()); + assertEquals(original.getEncryptedKeys(), deserialized.getEncryptedKeys()); + assertNotNull(deserialized.getFileIO()); } @Test @@ -227,11 +268,19 @@ public void testBeamSchemaCoderRoundtrip() throws Exception { assertEquals(original.getTableIdentifierString(), decoded.getTableIdentifierString()); assertEquals(original.getName(), decoded.getName()); assertEquals(original.getLocation(), decoded.getLocation()); + assertEquals(original.getSchemaId(), decoded.getSchemaId()); + assertEquals(original.getSchemasJson(), decoded.getSchemasJson()); assertEquals(original.getSpecId(), decoded.getSpecId()); + assertEquals(original.getPartitionSpecsJson(), decoded.getPartitionSpecsJson()); + assertEquals(original.getOrderId(), decoded.getOrderId()); + assertEquals(original.getSortOrdersJson(), decoded.getSortOrdersJson()); assertEquals(original.getSchema().asStruct(), decoded.getSchema().asStruct()); assertEquals(original.getPartitionSpec(), decoded.getPartitionSpec()); assertEquals(original.getSortOrder(), decoded.getSortOrder()); assertEquals(original.getProperties(), decoded.getProperties()); + assertEquals(original.getFileIoJson(), decoded.getFileIoJson()); + assertEquals(original.getEncryptedKeyJsons(), decoded.getEncryptedKeyJsons()); + assertNotNull(decoded.getFileIO()); } @Test @@ -257,14 +306,31 @@ public void testConcurrentGetterInitializationThreadSafety() throws Exception { () -> { startLatch.await(); Schema schema = spec.getSchema(); + Map schemas = spec.getSchemas(); PartitionSpec ps = spec.getPartitionSpec(); + Map specs = spec.getPartitionSpecs(); SortOrder so = spec.getSortOrder(); + Map orders = spec.getSortOrders(); TableIdentifier ti = spec.getTableIdentifier(); - - if (schema == null || ps == null || so == null || ti == null) { + FileIO io = spec.getFileIO(); + List keys = spec.getEncryptedKeys(); + + if (schema == null + || schemas == null + || ps == null + || specs == null + || so == null + || orders == null + || ti == null + || io == null + || keys == null) { throw new IllegalStateException("Getter returned null"); } - if (schema != spec.getSchema() || ps != spec.getPartitionSpec()) { + if (schemas != spec.getSchemas() + || specs != spec.getPartitionSpecs() + || orders != spec.getSortOrders() + || io != spec.getFileIO() + || keys != spec.getEncryptedKeys()) { throw new IllegalStateException("Getter returned non-identical instance"); } return null; diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java index 70da317051c6..663c818b5879 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java @@ -20,13 +20,10 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotEquals; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; -import java.util.Collections; import java.util.Map; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.CatalogProperties; @@ -68,13 +65,12 @@ public void setUp() throws Exception { @Test public void testConstructorNullChecks() { + assertThrows(NullPointerException.class, () -> new SideInputTable(null)); TableIdentifier tableId = TableIdentifier.of("default", "null_check_table"); Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA); SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); - - assertThrows(NullPointerException.class, () -> new SideInputTable(null, realTable.io())); - assertThrows(NullPointerException.class, () -> new SideInputTable(spec, null)); - assertThrows(NullPointerException.class, () -> new SideInputTable(spec, realTable.io(), null)); + assertThrows( + NullPointerException.class, () -> new SideInputTable(spec, (Map) null)); } @Test @@ -98,8 +94,7 @@ public void testMetadataDelegation() { .create(); SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); - SideInputTable sideInputTable = - new SideInputTable(spec, realTable.io(), PlaintextEncryptionManager.instance()); + SideInputTable sideInputTable = new SideInputTable(spec, ImmutableMap.of()); assertEquals(realTable.name(), sideInputTable.name()); assertEquals(realTable.location(), sideInputTable.location()); @@ -111,30 +106,13 @@ public void testMetadataDelegation() { assertEquals(realTable.sortOrders().keySet(), sideInputTable.sortOrders().keySet()); assertEquals( realTable.properties().get("user.key"), sideInputTable.properties().get("user.key")); - assertEquals(realTable.io(), sideInputTable.io()); + assertNotNull(sideInputTable.io()); + assertEquals(realTable.io().getClass().getName(), sideInputTable.io().getClass().getName()); assertNotNull(sideInputTable.locationProvider()); assertNotNull(sideInputTable.encryption()); + assertTrue(sideInputTable.encryption() instanceof PlaintextEncryptionManager); assertEquals(spec, sideInputTable.getTableSpec()); - assertTrue(sideInputTable.specs().containsKey(spec.getPartitionSpec().specId())); - - // Verify refresh is a safe no-op - sideInputTable.refresh(); - } - - @Test - public void testSnapshotQueriesReturnEmptyOrNull() { - TableIdentifier tableId = TableIdentifier.of("default", "snapshot_query_table"); - Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA); - SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); - SideInputTable sideInputTable = new SideInputTable(spec, realTable.io()); - - assertNull(sideInputTable.currentSnapshot()); - assertNull(sideInputTable.snapshot(12345L)); - assertEquals(Collections.emptyList(), ImmutableList.copyOf(sideInputTable.snapshots())); - assertEquals(Collections.emptyList(), sideInputTable.history()); - assertEquals(Collections.emptyMap(), sideInputTable.refs()); - assertEquals(Collections.emptyList(), sideInputTable.statisticsFiles()); - assertEquals(Collections.emptyList(), sideInputTable.partitionStatisticsFiles()); + assertTrue(sideInputTable.specs().containsKey(spec.getSpecId())); } @Test @@ -146,7 +124,7 @@ public void testWritingPartitionedWithRecordWriter() throws Exception { catalog.buildTable(tableId, TestFixtures.SCHEMA).withPartitionSpec(partitionSpec).create(); SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); - SideInputTable sideInputTable = new SideInputTable(spec, realTable.io()); + SideInputTable sideInputTable = new SideInputTable(spec); PartitionKey partitionKey = new PartitionKey(sideInputTable.spec(), sideInputTable.schema()); Record record = GenericRecord.create(sideInputTable.schema()); @@ -173,7 +151,7 @@ public void testWritingUnpartitionedWithRecordWriter() throws Exception { Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA); SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); - SideInputTable sideInputTable = new SideInputTable(spec, realTable.io()); + SideInputTable sideInputTable = new SideInputTable(spec); PartitionKey partitionKey = new PartitionKey(sideInputTable.spec(), sideInputTable.schema()); Record record = GenericRecord.create(sideInputTable.schema()); @@ -199,12 +177,23 @@ public void testWritingUnpartitionedWithRecordWriter() throws Exception { } @Test - public void testUnsupportedOperationsThrowExceptions() { + public void testUnsupportedAndNoOpOperationsThrowExceptions() { TableIdentifier tableId = TableIdentifier.of("default", "mutations_test_table"); Table realTable = catalog.createTable(tableId, TestFixtures.SCHEMA); SerializableTableSpec spec = SerializableTableSpec.fromTable(tableId, realTable); - SideInputTable sideInputTable = new SideInputTable(spec, realTable.io()); - + SideInputTable sideInputTable = new SideInputTable(spec); + + // Refresh & Snapshot operations must throw UnsupportedOperationException + assertThrows(UnsupportedOperationException.class, sideInputTable::refresh); + assertThrows(UnsupportedOperationException.class, sideInputTable::currentSnapshot); + assertThrows(UnsupportedOperationException.class, () -> sideInputTable.snapshot(12345L)); + assertThrows(UnsupportedOperationException.class, sideInputTable::snapshots); + assertThrows(UnsupportedOperationException.class, sideInputTable::history); + assertThrows(UnsupportedOperationException.class, sideInputTable::refs); + assertThrows(UnsupportedOperationException.class, sideInputTable::statisticsFiles); + assertThrows(UnsupportedOperationException.class, sideInputTable::partitionStatisticsFiles); + + // Scans & Mutations assertThrows(UnsupportedOperationException.class, sideInputTable::newScan); assertThrows(UnsupportedOperationException.class, sideInputTable::newIncrementalAppendScan); assertThrows(UnsupportedOperationException.class, sideInputTable::newIncrementalChangelogScan); @@ -237,9 +226,9 @@ public void testEqualsHashCodeAndToString() { SerializableTableSpec spec1 = SerializableTableSpec.fromTable(tableId1, realTable1); SerializableTableSpec spec2 = SerializableTableSpec.fromTable(tableId2, realTable2); - SideInputTable table1a = new SideInputTable(spec1, realTable1.io()); - SideInputTable table1b = new SideInputTable(spec1, realTable1.io()); - SideInputTable table2 = new SideInputTable(spec2, realTable2.io()); + SideInputTable table1a = new SideInputTable(spec1); + SideInputTable table1b = new SideInputTable(spec1); + SideInputTable table2 = new SideInputTable(spec2); assertEquals(table1a, table1b); assertEquals(table1a.hashCode(), table1b.hashCode());