[SPARK-51372][SQL] Introduce TableInfo for table creations

### What changes were proposed in this pull request?

Introduces a new `TableInfo`  for operations in `TableCatalog`.

A followup PR will be sent for the table replace methods in `StagingTableCatalog`.

### Why are the changes needed?

The current `TableCatalog` interface has overloaded `createTable()` implementations. As we are adding new parameters, we have to deprecate the old ones. This PR introduces the `TableInfo` POJO so that the future code changes will be cleaner.

### Does this PR introduce _any_ user-facing change?

Yes.  But the PR is backwards compatible by providing a default implementation.

### How was this patch tested?

* New unit tests.
* Existing tests in `DataSourceV2Suite`.

### Was this patch authored or co-authored using generative AI tooling?

No

Closes #50137 from anoopj/anoopj-builder.

Lead-authored-by: Anoop Johnson <[email protected]>
Co-authored-by: Anoop Johnson <[email protected]>
Co-authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>

Author: anoopj

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableCatalog.java

@@ -33,11 +33,11 @@
 /**
  * Catalog methods for working with Tables.
  * <p>
- * TableCatalog implementations may be case sensitive or case insensitive. Spark will pass
+ * TableCatalog implementations may be case-sensitive or case-insensitive. Spark will pass
  * {@link Identifier table identifiers} without modification. Field names passed to
  * {@link #alterTable(Identifier, TableChange...)} will be normalized to match the case used in the
- * table schema when updating, renaming, or dropping existing columns when catalyst analysis is case
- * insensitive.
+ * table schema when updating, renaming, or dropping existing columns when catalyst analysis is
+ * case-insensitive.
  *
  * @since 3.0.0
  */
@@ -208,26 +208,37 @@ default Table createTable(
     throw QueryCompilationErrors.mustOverrideOneMethodError("createTable");
   }
 
+  /**
+   * Create a table in the catalog.
+   * <p>
+   * @deprecated This is deprecated. Please override
+   * {@link #createTable(Identifier, TableInfo)} instead.
+   */
+  @Deprecated(since = "4.1.0")
+  default Table createTable(
+      Identifier ident,
+      Column[] columns,
+      Transform[] partitions,
+      Map<String, String> properties) throws TableAlreadyExistsException, NoSuchNamespaceException {
+    return createTable(ident, CatalogV2Util.v2ColumnsToStructType(columns), partitions, properties);
+  }
+
   /**
    * Create a table in the catalog.
    *
    * @param ident a table identifier
-   * @param columns the columns of the new table.
-   * @param partitions transforms to use for partitioning data in the table
-   * @param properties a string map of table properties
+   * @param tableInfo information about the table.
    * @return metadata for the new table. This can be null if getting the metadata for the new table
    *         is expensive. Spark will call {@link #loadTable(Identifier)} if needed (e.g. CTAS).
    *
    * @throws TableAlreadyExistsException If a table or view already exists for the identifier
    * @throws UnsupportedOperationException If a requested partition transform is not supported
    * @throws NoSuchNamespaceException If the identifier namespace does not exist (optional)
+   * @since 4.1.0
    */
-  default Table createTable(
-      Identifier ident,
-      Column[] columns,
-      Transform[] partitions,
-      Map<String, String> properties) throws TableAlreadyExistsException, NoSuchNamespaceException {
-    return createTable(ident, CatalogV2Util.v2ColumnsToStructType(columns), partitions, properties);
+  default Table createTable(Identifier ident, TableInfo tableInfo)
+      throws TableAlreadyExistsException, NoSuchNamespaceException {
+    return createTable(ident, tableInfo.columns(), tableInfo.partitions(), tableInfo.properties());
   }
 
   /**

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableInfo.java

@@ -0,0 +1,85 @@
+/*
+ * 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.spark.sql.connector.catalog;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+import com.google.common.collect.Maps;
+import org.apache.spark.sql.connector.expressions.Transform;
+import org.apache.spark.sql.types.StructType;
+
+import java.util.Collections;
+import java.util.Map;
+
+public class TableInfo {
+
+  private final Column[] columns;
+  private final Map<String, String> properties;
+  private final Transform[] partitions;
+
+  /**
+   * Constructor for TableInfo used by the builder.
+   * @param builder Builder.
+   */
+  private TableInfo(Builder builder) {
+    this.columns = builder.columns;
+    this.properties = Collections.unmodifiableMap(builder.properties);
+    this.partitions = builder.partitions;
+  }
+
+  public Column[] columns() {
+    return columns;
+  }
+
+  public StructType schema() {
+    return CatalogV2Util.v2ColumnsToStructType(columns);
+  }
+
+  public Map<String, String> properties() {
+    return properties;
+  }
+
+  public Transform[] partitions() {
+    return partitions;
+  }
+
+  public static class Builder {
+    private Column[] columns;
+    private Map<String, String> properties;
+    private Transform[] partitions;
+
+    public Builder withColumns(Column[] columns) {
+      this.columns = columns;
+      return this;
+    }
+
+    public Builder withProperties(Map<String, String> properties) {
+      this.properties = Maps.newHashMap();
+      this.properties.putAll(properties);
+      return this;
+    }
+
+    public Builder withPartitions(Transform[] partitions) {
+      this.partitions = partitions;
+      return this;
+    }
+
+    public TableInfo build() {
+      checkNotNull(columns, "columns should not be null");
+      return new TableInfo(this);
+    }
+  }
+}