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..c6ee4a976993 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpec.java @@ -0,0 +1,382 @@ +/* + * 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.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; +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.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; +import org.apache.iceberg.SchemaParser; +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 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 +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 getSchemaId(); + + @SchemaFieldNumber("4") + public abstract Map getSchemasJson(); + + @SchemaFieldNumber("5") + public abstract int getSpecId(); + + @SchemaFieldNumber("6") + public abstract Map getPartitionSpecsJson(); + + @SchemaFieldNumber("7") + public abstract int getOrderId(); + + @SchemaFieldNumber("8") + public abstract Map getSortOrdersJson(); + + @SchemaFieldNumber("9") + public abstract Map getProperties(); + + @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 Map getSchemas() { + Map local = cachedSchemas; + if (local == null) { + synchronized (this) { + local = cachedSchemas; + if (local == null) { + ImmutableMap.Builder builder = ImmutableMap.builder(); + for (Map.Entry entry : getSchemasJson().entrySet()) { + builder.put(entry.getKey(), SchemaParser.fromJson(entry.getValue())); + } + cachedSchemas = local = builder.build(); + } + } + } + return local; + } + + @SchemaIgnore + 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 = cachedPartitionSpecs; + if (local == null) { + ImmutableMap.Builder builder = ImmutableMap.builder(); + for (Map.Entry entry : getPartitionSpecsJson().entrySet()) { + builder.put( + entry.getKey(), PartitionSpecParser.fromJson(getSchema(), entry.getValue())); + } + cachedPartitionSpecs = local = builder.build(); + } + } + } + return local; + } + + @SchemaIgnore + 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 = cachedSortOrders; + if (local == null) { + 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; + if (local == null) { + synchronized (this) { + local = cachedTableIdentifier; + if (local == null) { + cachedTableIdentifier = + local = IcebergUtils.parseTableIdentifier(getTableIdentifierString()); + } + } + } + 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(); + } + + 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 setSchemaId(int schemaId); + + public abstract Builder setSchemasJson(Map schemasJson); + + public abstract Builder setSpecId(int specId); + + public abstract Builder setPartitionSpecsJson(Map partitionSpecsJson); + + public abstract Builder setOrderId(int orderId); + + 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(); + } + + /** + * 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) { + 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()) + .setPartitionSpecsJson(specsJson.build()) + .setOrderId(table.sortOrder().orderId()) + .setSortOrdersJson(sortOrdersJson.build()) + .setProperties(table.properties()) + .setFileIoJson(FileIOParser.toJson(table.io())) + .setEncryptedKeyJsons(encryptedKeyJsons) + .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..aa51571c4cbd --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SideInputTable.java @@ -0,0 +1,368 @@ +/* + * 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.TableProperties; +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.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; + +/** + * A lightweight adapter that implements {@link Table} backed by a {@link SerializableTableSpec}. + * + *

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}. + * + *

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 EncryptionManager encryptionManager; + private final LocationProvider locationProvider; + + public SideInputTable(SerializableTableSpec spec) { + this(spec, Collections.emptyMap()); + } + + 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.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 spec.getSchemas(); + } + + @Override + public PartitionSpec spec() { + return spec.getPartitionSpec(); + } + + @Override + public Map specs() { + return spec.getPartitionSpecs(); + } + + @Override + public SortOrder sortOrder() { + return spec.getSortOrder(); + } + + @Override + public Map sortOrders() { + return spec.getSortOrders(); + } + + @Override + public Map properties() { + return spec.getProperties(); + } + + @Override + public LocationProvider locationProvider() { + return locationProvider; + } + + @Override + public FileIO io() { + return spec.getFileIO(); + } + + @Override + public EncryptionManager encryption() { + return encryptionManager; + } + + @Override + public void refresh() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support refresh."); + } + + @Override + public Snapshot currentSnapshot() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support snapshots."); + } + + @Override + public Snapshot snapshot(long snapshotId) { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support snapshots."); + } + + @Override + public Iterable snapshots() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support snapshots."); + } + + @Override + public List history() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support snapshots."); + } + + @Override + public Map refs() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support snapshot refs."); + } + + @Override + public List statisticsFiles() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support statisticsFiles."); + } + + @Override + public List partitionStatisticsFiles() { + throw new UnsupportedOperationException( + "SideInputTable is a read-only metadata adapter and does not support partitionStatisticsFiles."); + } + + @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(encryptionManager, that.encryptionManager); + } + + @Override + public int hashCode() { + return Objects.hash(spec, encryptionManager); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("spec", spec) + .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..87a843db7a2f --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SerializableTableSpecTest.java @@ -0,0 +1,348 @@ +/* + * 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.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; +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.encryption.EncryptedKey; +import org.apache.iceberg.io.FileIO; +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 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"); + 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.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 + 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(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 + 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()); + assertNotNull(spec.getFileIO()); + } + + @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.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 + 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.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 + @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(); + Map schemas = spec.getSchemas(); + PartitionSpec ps = spec.getPartitionSpec(); + Map specs = spec.getPartitionSpecs(); + SortOrder so = spec.getSortOrder(); + Map orders = spec.getSortOrders(); + TableIdentifier ti = spec.getTableIdentifier(); + 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 (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; + })); + } + + 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..663c818b5879 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SideInputTableTest.java @@ -0,0 +1,239 @@ +/* + * 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.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.util.Map; +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() { + 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(spec, (Map) 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, ImmutableMap.of()); + + 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")); + 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.getSpecId())); + } + + @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); + + 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); + + 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 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); + + // 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); + 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); + SideInputTable table1b = new SideInputTable(spec1); + SideInputTable table2 = new SideInputTable(spec2); + + assertEquals(table1a, table1b); + assertEquals(table1a.hashCode(), table1b.hashCode()); + assertNotEquals(table1a, table2); + assertNotNull(table1a.toString()); + assertTrue(table1a.toString().contains("SideInputTable")); + } +}