Support for file paths in SparkCatalogs via HadoopTables

core/src/main/java/org/apache/iceberg/hadoop/HadoopTables.java

@@ -37,6 +37,9 @@
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.Tables;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.Transactions;
+import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
 import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
@@ -131,20 +134,10 @@ private Table loadMetadataTable(String location, String metadataTableName, Metad
   @Override
   public Table create(Schema schema, PartitionSpec spec, SortOrder order,
                       Map<String, String> properties, String location) {
-    Preconditions.checkNotNull(schema, "A table schema is required");
-
-    TableOperations ops = newTableOps(location);
-    if (ops.current() != null) {
-      throw new AlreadyExistsException("Table already exists at location: %s", location);
-    }
-
-    Map<String, String> tableProps = properties == null ? ImmutableMap.of() : properties;
-    PartitionSpec partitionSpec = spec == null ? PartitionSpec.unpartitioned() : spec;
-    SortOrder sortOrder = order == null ? SortOrder.unsorted() : order;
-    TableMetadata metadata = TableMetadata.newTableMetadata(schema, partitionSpec, sortOrder, location, tableProps);
-    ops.commit(null, metadata);
-
-    return new BaseTable(ops, location);
+    return buildTable(location, schema).withPartitionSpec(spec)
+        .withSortOrder(order)
+        .withProperties(properties)
+        .create();
   }
 
   /**
@@ -200,6 +193,165 @@ TableOperations newTableOps(String location) {
     }
   }
 
+  private TableMetadata tableMetadata(Schema schema, PartitionSpec spec, SortOrder order,
+                                      Map<String, String> properties, String location) {
+    Preconditions.checkNotNull(schema, "A table schema is required");
+
+    Map<String, String> tableProps = properties == null ? ImmutableMap.of() : properties;
+    PartitionSpec partitionSpec = spec == null ? PartitionSpec.unpartitioned() : spec;
+    SortOrder sortOrder = order == null ? SortOrder.unsorted() : order;
+    return TableMetadata.newTableMetadata(schema, partitionSpec, sortOrder, location, tableProps);
+  }
+
+  /**
+   * Start a transaction to create a table.
+   *
+   * @param location a location for the table
+   * @param schema a schema
+   * @param spec a partition spec
+   * @param properties a string map of table properties
+   * @return a {@link Transaction} to create the table
+   * @throws AlreadyExistsException if the table already exists
+   */
+  public Transaction newCreateTableTransaction(
+      String location,
+      Schema schema,
+      PartitionSpec spec,
+      Map<String, String> properties) {
+    return buildTable(location, schema).withPartitionSpec(spec).withProperties(properties).createTransaction();
+  }
+
+  /**
+   * Start a transaction to replace a table.
+   *
+   * @param location a location for the table
+   * @param schema a schema
+   * @param spec a partition spec
+   * @param properties a string map of table properties
+   * @param orCreate whether to create the table if not exists
+   * @return a {@link Transaction} to replace the table
+   * @throws NoSuchTableException if the table doesn't exist and orCreate is false
+   */
+  public Transaction newReplaceTableTransaction(
+      String location,
+      Schema schema,
+      PartitionSpec spec,
+      Map<String, String> properties,
+      boolean orCreate) {
+
+
+    Catalog.TableBuilder builder = buildTable(location, schema).withPartitionSpec(spec).withProperties(properties);
+    return orCreate ? builder.createOrReplaceTransaction() : builder.replaceTransaction();
+  }
+
+  public Catalog.TableBuilder buildTable(String location, Schema schema) {
+    return new HadoopTableBuilder(location, schema);
+  }
+
+  private class HadoopTableBuilder implements Catalog.TableBuilder {
+    private final String location;
+    private final Schema schema;
+    private final ImmutableMap.Builder<String, String> propertiesBuilder = ImmutableMap.builder();
+    private PartitionSpec spec = PartitionSpec.unpartitioned();
+    private SortOrder sortOrder = SortOrder.unsorted();
+
+
+    HadoopTableBuilder(String location, Schema schema) {
+      this.location = location;
+      this.schema = schema;
+    }
+
+    @Override
+    public Catalog.TableBuilder withPartitionSpec(PartitionSpec newSpec) {
+      this.spec = newSpec != null ? newSpec : PartitionSpec.unpartitioned();
+      return this;
+    }
+
+    @Override
+    public Catalog.TableBuilder withSortOrder(SortOrder newSortOrder) {
+      this.sortOrder = newSortOrder != null ? newSortOrder : SortOrder.unsorted();
+      return this;
+    }
+
+    @Override
+    public Catalog.TableBuilder withLocation(String newLocation) {
+      Preconditions.checkArgument(newLocation == null || location.equals(newLocation),
+          String.format("Table location %s differs from the table location (%s) from the PathIdentifier",
+              newLocation, location));
+      return this;
+    }
+
+    @Override
+    public Catalog.TableBuilder withProperties(Map<String, String> properties) {
+      if (properties != null) {
+        propertiesBuilder.putAll(properties);
+      }
+      return this;
+    }
+
+    @Override
+    public Catalog.TableBuilder withProperty(String key, String value) {
+      propertiesBuilder.put(key, value);
+      return this;
+    }
+
+    @Override
+    public Table create() {
+      TableOperations ops = newTableOps(location);
+      if (ops.current() != null) {
+        throw new AlreadyExistsException("Table already exists at location: %s", location);
+      }
+
+      Map<String, String> properties = propertiesBuilder.build();
+      TableMetadata metadata = tableMetadata(schema, spec, sortOrder, properties, location);
+      ops.commit(null, metadata);
+      return new BaseTable(ops, location);
+    }
+
+    @Override
+    public Transaction createTransaction() {
+      TableOperations ops = newTableOps(location);
+      if (ops.current() != null) {
+        throw new AlreadyExistsException("Table already exists: %s", location);
+      }
+
+      Map<String, String> properties = propertiesBuilder.build();
+      TableMetadata metadata = tableMetadata(schema, spec, null, properties, location);
+      return Transactions.createTableTransaction(location, ops, metadata);
+    }
+
+    @Override
+    public Transaction replaceTransaction() {
+      return newReplaceTableTransaction(false);
+    }
+
+    @Override
+    public Transaction createOrReplaceTransaction() {
+      return newReplaceTableTransaction(true);
+    }
+
+    private Transaction newReplaceTableTransaction(boolean orCreate) {
+      TableOperations ops = newTableOps(location);
+      if (!orCreate && ops.current() == null) {
+        throw new NoSuchTableException("No such table: %s", location);
+      }
+
+      Map<String, String> properties = propertiesBuilder.build();
+      TableMetadata metadata;
+      if (ops.current() != null) {
+        metadata = ops.current().buildReplacement(schema, spec, SortOrder.unsorted(), location, properties);
+      } else {
+        metadata = tableMetadata(schema, spec, null, properties, location);
+      }
+
+      if (orCreate) {
+        return Transactions.createOrReplaceTableTransaction(location, ops, metadata);
+      } else {
+        return Transactions.replaceTableTransaction(location, ops, metadata);
+      }
+    }
+  }
+
   @Override
   public void setConf(Configuration conf) {
     this.conf = conf;

spark3/src/main/java/org/apache/iceberg/spark/PathIdentifier.java

@@ -0,0 +1,57 @@
+/*
+ * 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.iceberg.spark;
+
+import java.util.List;
+import org.apache.iceberg.relocated.com.google.common.base.Joiner;
+import org.apache.iceberg.relocated.com.google.common.base.Splitter;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.spark.sql.connector.catalog.Identifier;
+
+public class PathIdentifier implements Identifier {
+  private static final Splitter SPLIT = Splitter.on("/");
+  private static final Joiner JOIN = Joiner.on("/");
+  private final String[] namespace;
+  private final String location;
+  private final String name;
+
+  public PathIdentifier(String location) {
+    this.location = location;
+    List<String> pathParts = SPLIT.splitToList(location);
+    name = Iterables.getLast(pathParts);
+    namespace = pathParts.size() > 1 ?
+        new String[]{JOIN.join(pathParts.subList(0, pathParts.size() - 1))} :
+        new String[0];
+  }
+
+  @Override
+  public String[] namespace() {
+    return namespace;
+  }
+
+  @Override
+  public String name() {
+    return name;
+  }
+
+  public String location() {
+    return location;
+  }
+}