Revert "Add support for Iceberg table identifiers with special characters (#33293)"

This reverts commit d6e0b0c.

Author: Abacn

.github/trigger_files/IO_Iceberg_Integration_Tests.json

@@ -55,10 +55,8 @@ dependencies {
     implementation "org.apache.iceberg:iceberg-api:$iceberg_version"
     implementation "org.apache.iceberg:iceberg-parquet:$iceberg_version"
     implementation "org.apache.iceberg:iceberg-orc:$iceberg_version"
-    runtimeOnly "org.apache.iceberg:iceberg-gcp:$iceberg_version"
     implementation library.java.hadoop_common
-    implementation library.java.jackson_core
-    implementation library.java.jackson_databind
+    runtimeOnly "org.apache.iceberg:iceberg-gcp:$iceberg_version"
 
     testImplementation project(":sdks:java:managed")
     testImplementation library.java.hadoop_client

sdks/java/io/iceberg/build.gradle

@@ -47,6 +47,7 @@
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.io.FileIO;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.slf4j.Logger;
@@ -132,7 +133,7 @@ public void processElement(
         return;
       }
 
-      Table table = getCatalog().loadTable(IcebergUtils.parseTableIdentifier(element.getKey()));
+      Table table = getCatalog().loadTable(TableIdentifier.parse(element.getKey()));
 
       // vast majority of the time, we will simply append data files.
       // in the rare case we get a batch that contains multiple partition specs, we will group

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AppendFilesToTables.java

@@ -25,7 +25,6 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.catalog.TableIdentifierParser;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 
 @AutoValue
@@ -42,7 +41,7 @@ abstract class FileWriteResult {
   @SchemaIgnore
   public TableIdentifier getTableIdentifier() {
     if (cachedTableIdentifier == null) {
-      cachedTableIdentifier = IcebergUtils.parseTableIdentifier(getTableIdentifierString());
+      cachedTableIdentifier = TableIdentifier.parse(getTableIdentifierString());
     }
     return cachedTableIdentifier;
   }
@@ -68,7 +67,7 @@ abstract static class Builder {
 
     @SchemaIgnore
     public Builder setTableIdentifier(TableIdentifier tableId) {
-      return setTableIdentifierString(TableIdentifierParser.toJson(tableId));
+      return setTableIdentifierString(tableId.toString());
     }
 
     public abstract FileWriteResult build();

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/FileWriteResult.java

@@ -31,6 +31,7 @@
 import org.apache.beam.sdk.values.PCollection;
 import org.apache.beam.sdk.values.PCollectionRowTuple;
 import org.apache.beam.sdk.values.Row;
+import org.apache.iceberg.catalog.TableIdentifier;
 
 /**
  * SchemaTransform implementation for {@link IcebergIO#readRows}. Reads records from Iceberg and
@@ -85,7 +86,7 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
               .getPipeline()
               .apply(
                   IcebergIO.readRows(configuration.getIcebergCatalog())
-                      .from(IcebergUtils.parseTableIdentifier(configuration.getTable())));
+                      .from(TableIdentifier.parse(configuration.getTable())));
 
       return PCollectionRowTuple.of(OUTPUT_TAG, output);
     }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergReadSchemaTransformProvider.java

@@ -23,7 +23,6 @@
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.catalog.TableIdentifierParser;
 import org.apache.iceberg.expressions.Expression;
 import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -52,9 +51,7 @@ public enum ScanType {
   public Table getTable() {
     if (cachedTable == null) {
       cachedTable =
-          getCatalogConfig()
-              .catalog()
-              .loadTable(IcebergUtils.parseTableIdentifier(getTableIdentifier()));
+          getCatalogConfig().catalog().loadTable(TableIdentifier.parse(getTableIdentifier()));
     }
     return cachedTable;
   }
@@ -129,7 +126,7 @@ public abstract static class Builder {
     public abstract Builder setTableIdentifier(String tableIdentifier);
 
     public Builder setTableIdentifier(TableIdentifier tableIdentifier) {
-      return this.setTableIdentifier(TableIdentifierParser.toJson(tableIdentifier));
+      return this.setTableIdentifier(tableIdentifier.toString());
     }
 
     public Builder setTableIdentifier(String... names) {

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergScanConfig.java

@@ -19,9 +19,6 @@
 
 import static org.apache.beam.sdk.util.Preconditions.checkArgumentNotNull;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import java.nio.ByteBuffer;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -39,8 +36,6 @@
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
 import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
-import org.apache.iceberg.catalog.TableIdentifier;
-import org.apache.iceberg.catalog.TableIdentifierParser;
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.types.Type;
@@ -52,9 +47,6 @@
 
 /** Utilities for converting between Beam and Iceberg types, made public for user's convenience. */
 public class IcebergUtils {
-
-  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
-
   private IcebergUtils() {}
 
   private static final Map<Schema.TypeName, Type> BEAM_TYPES_TO_ICEBERG_TYPES =
@@ -514,13 +506,4 @@ private static Object getLogicalTypeValue(Object icebergValue, Schema.FieldType
     // LocalDateTime, LocalDate, LocalTime
     return icebergValue;
   }
-
-  public static TableIdentifier parseTableIdentifier(String table) {
-    try {
-      JsonNode jsonNode = OBJECT_MAPPER.readTree(table);
-      return TableIdentifierParser.fromJson(jsonNode);
-    } catch (JsonProcessingException e) {
-      return TableIdentifier.parse(table);
-    }
-  }
 }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java

@@ -41,7 +41,7 @@ class OneTableDynamicDestinations implements DynamicDestinations, Externalizable
   @VisibleForTesting
   TableIdentifier getTableIdentifier() {
     if (tableId == null) {
-      tableId = IcebergUtils.parseTableIdentifier(checkStateNotNull(tableIdString));
+      tableId = TableIdentifier.parse(checkStateNotNull(tableIdString));
     }
     return tableId;
   }
@@ -86,6 +86,6 @@ public void writeExternal(ObjectOutput out) throws IOException {
   @Override
   public void readExternal(ObjectInput in) throws IOException {
     tableIdString = in.readUTF();
-    tableId = IcebergUtils.parseTableIdentifier(tableIdString);
+    tableId = TableIdentifier.parse(tableIdString);
   }
 }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/OneTableDynamicDestinations.java

@@ -24,6 +24,7 @@
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.sdk.values.ValueInSingleWindow;
 import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.catalog.TableIdentifier;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 class PortableIcebergDestinations implements DynamicDestinations {
@@ -72,7 +73,7 @@ public String getTableStringIdentifier(ValueInSingleWindow<Row> element) {
   @Override
   public IcebergDestination instantiateDestination(String dest) {
     return IcebergDestination.builder()
-        .setTableIdentifier(IcebergUtils.parseTableIdentifier(dest))
+        .setTableIdentifier(TableIdentifier.parse(dest))
         .setTableCreateConfig(null)
         .setFileFormat(FileFormat.fromString(fileFormat))
         .build();

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PortableIcebergDestinations.java

@@ -23,8 +23,6 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.util.Arrays;
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -70,11 +68,11 @@
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
+import org.junit.runners.JUnit4;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-@RunWith(Parameterized.class)
+@RunWith(JUnit4.class)
 public class IcebergIOReadTest {
 
   private static final Logger LOG = LoggerFactory.getLogger(IcebergIOReadTest.class);
@@ -85,21 +83,6 @@ public class IcebergIOReadTest {
 
   @Rule public TestPipeline testPipeline = TestPipeline.create();
 
-  @Parameterized.Parameters
-  public static Collection<Object[]> data() {
-    return Arrays.asList(
-        new Object[][] {
-          {String.format("{\"namespace\": [\"default\"], \"name\": \"%s\"}", tableId())},
-          {String.format("default.%s", tableId())},
-        });
-  }
-
-  public static String tableId() {
-    return "table" + Long.toString(UUID.randomUUID().hashCode(), 16);
-  }
-
-  @Parameterized.Parameter public String tableStringIdentifier;
-
   static class PrintRow extends DoFn<Row, Row> {
 
     @ProcessElement
@@ -111,7 +94,8 @@ public void process(@Element Row row, OutputReceiver<Row> output) throws Excepti
 
   @Test
   public void testSimpleScan() throws Exception {
-    TableIdentifier tableId = IcebergUtils.parseTableIdentifier(tableStringIdentifier);
+    TableIdentifier tableId =
+        TableIdentifier.of("default", "table" + Long.toString(UUID.randomUUID().hashCode(), 16));
     Table simpleTable = warehouse.createTable(tableId, TestFixtures.SCHEMA);
     final Schema schema = IcebergUtils.icebergSchemaToBeamSchema(TestFixtures.SCHEMA);