HIVE-29694: Reuse DB column descriptors when adding partitions in HMS

ql/src/test/queries/clientpositive/partition_descriptor_reuse.q

@@ -0,0 +1,16 @@
+-- Debug instructions:
+-- 1. Run the test in debug mode: mvn test -Dtest=TestMiniLlapLocalCliDriver -Dqfile=partition_descriptor_reuse.q -Dtest.output.overwrite -Dtest.metastore.db=postgres -Dmaven.surefire.debug
+-- 2. Put a breakpoint in CliAdapter.java:120 CliAdapter.this.tearDown() to prevent DB from being dropped.
+-- 3. Connect to the dockerized database and check the state of the metadata
+-- docker exec -it CONTAINER_NAME psql -U test -d hivedb
+-- TODO: Probably not gonna need this file at the end so drop it
+create table person(id string, fname string) partitioned by (birthyear string);
+alter table person add partition (birthyear='1987');
+alter table person add partition (birthyear='1988');
+alter table person add partition (birthyear='1989');
+alter table person add partition (birthyear='1990');
+alter table person add partition (birthyear='1991');
+
+alter table person partition (birthyear='1989') add columns (lname string);
+alter table person partition (birthyear='1990') add columns (lname string);
+alter table person partition (birthyear='1991') change column fname fullname string;

ql/src/test/results/clientpositive/llap/partition_descriptor_reuse.q.out

@@ -0,0 +1,70 @@
+PREHOOK: query: create table person(id string, fname string) partitioned by (birthyear string)
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@person
+POSTHOOK: query: create table person(id string, fname string) partitioned by (birthyear string)
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@person
+PREHOOK: query: alter table person add partition (birthyear='1987')
+PREHOOK: type: ALTERTABLE_ADDPARTS
+PREHOOK: Output: default@person
+POSTHOOK: query: alter table person add partition (birthyear='1987')
+POSTHOOK: type: ALTERTABLE_ADDPARTS
+POSTHOOK: Output: default@person
+POSTHOOK: Output: default@person@birthyear=1987
+PREHOOK: query: alter table person add partition (birthyear='1988')
+PREHOOK: type: ALTERTABLE_ADDPARTS
+PREHOOK: Output: default@person
+POSTHOOK: query: alter table person add partition (birthyear='1988')
+POSTHOOK: type: ALTERTABLE_ADDPARTS
+POSTHOOK: Output: default@person
+POSTHOOK: Output: default@person@birthyear=1988
+PREHOOK: query: alter table person add partition (birthyear='1989')
+PREHOOK: type: ALTERTABLE_ADDPARTS
+PREHOOK: Output: default@person
+POSTHOOK: query: alter table person add partition (birthyear='1989')
+POSTHOOK: type: ALTERTABLE_ADDPARTS
+POSTHOOK: Output: default@person
+POSTHOOK: Output: default@person@birthyear=1989
+PREHOOK: query: alter table person add partition (birthyear='1990')
+PREHOOK: type: ALTERTABLE_ADDPARTS
+PREHOOK: Output: default@person
+POSTHOOK: query: alter table person add partition (birthyear='1990')
+POSTHOOK: type: ALTERTABLE_ADDPARTS
+POSTHOOK: Output: default@person
+POSTHOOK: Output: default@person@birthyear=1990
+PREHOOK: query: alter table person add partition (birthyear='1991')
+PREHOOK: type: ALTERTABLE_ADDPARTS
+PREHOOK: Output: default@person
+POSTHOOK: query: alter table person add partition (birthyear='1991')
+POSTHOOK: type: ALTERTABLE_ADDPARTS
+POSTHOOK: Output: default@person
+POSTHOOK: Output: default@person@birthyear=1991
+PREHOOK: query: alter table person partition (birthyear='1989') add columns (lname string)
+PREHOOK: type: ALTERTABLE_ADDCOLS
+PREHOOK: Input: default@person
+PREHOOK: Output: default@person@birthyear=1989
+POSTHOOK: query: alter table person partition (birthyear='1989') add columns (lname string)
+POSTHOOK: type: ALTERTABLE_ADDCOLS
+POSTHOOK: Input: default@person
+POSTHOOK: Input: default@person@birthyear=1989
+POSTHOOK: Output: default@person@birthyear=1989
+PREHOOK: query: alter table person partition (birthyear='1990') add columns (lname string)
+PREHOOK: type: ALTERTABLE_ADDCOLS
+PREHOOK: Input: default@person
+PREHOOK: Output: default@person@birthyear=1990
+POSTHOOK: query: alter table person partition (birthyear='1990') add columns (lname string)
+POSTHOOK: type: ALTERTABLE_ADDCOLS
+POSTHOOK: Input: default@person
+POSTHOOK: Input: default@person@birthyear=1990
+POSTHOOK: Output: default@person@birthyear=1990
+PREHOOK: query: alter table person partition (birthyear='1991') change column fname fullname string
+PREHOOK: type: ALTERTABLE_RENAMECOL
+PREHOOK: Input: default@person
+PREHOOK: Output: default@person@birthyear=1991
+POSTHOOK: query: alter table person partition (birthyear='1991') change column fname fullname string
+POSTHOOK: type: ALTERTABLE_RENAMECOL
+POSTHOOK: Input: default@person
+POSTHOOK: Input: default@person@birthyear=1991
+POSTHOOK: Output: default@person@birthyear=1991

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ColumnDescriptorSupplier.java

@@ -0,0 +1,38 @@
+/*
+ * 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.hadoop.hive.metastore;
+
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.model.MColumnDescriptor;
+import org.apache.hadoop.hive.metastore.model.MTable;
+
+import java.util.List;
+
+public interface ColumnDescriptorSupplier {
+  /**
+   * Gets a column descriptor for the provided table and list of columns. If no matching column descriptor exists,
+   * returns null. The column descriptor is considered a match if it has the same number of columns, and each column has
+   * the same name, type, and comment in the same order.
+   *
+   * @param cols the columns to match
+   * @param mt the table to search
+   * @return an existing MColumnDescriptor if found, null otherwise
+   */
+  MColumnDescriptor getColumnDescriptor(List<FieldSchema> cols, MTable mt) throws MetaException;
+}

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java

@@ -258,7 +258,7 @@
  * to be made into a interface that can read both from a database and a
  * filestore.
  */
-public class ObjectStore implements RawStore, Configurable {
+public class ObjectStore implements RawStore, ColumnDescriptorSupplier, Configurable {
   protected int batchSize = NO_BATCHING;
 
   private static final DateTimeFormatter YMDHMS_FORMAT = DateTimeFormatter.ofPattern(
@@ -1821,11 +1821,11 @@ private static MSerDeInfo convertToMSerDeInfo(SerDeInfo ms) throws MetaException
    * @param cols the columns the column descriptor contains
    * @return a new column descriptor db-backed object
    */
-  private static MColumnDescriptor createNewMColumnDescriptor(List<MFieldSchema> cols) {
+  private static MColumnDescriptor createNewMColumnDescriptor(List<FieldSchema> cols) {
     if (cols == null) {
       return null;
     }
-    return new MColumnDescriptor(cols);
+    return new MColumnDescriptor(convertToMFieldSchemas(cols));
   }
 
   private static StorageDescriptor convertToStorageDescriptor(
@@ -1929,7 +1929,28 @@ private static MStorageDescriptor convertToMStorageDescriptor(StorageDescriptor
     if (sd == null) {
       return null;
     }
-    MColumnDescriptor mcd = createNewMColumnDescriptor(convertToMFieldSchemas(sd.getCols()));
+    MColumnDescriptor mcd = createNewMColumnDescriptor(sd.getCols());
+    return convertToMStorageDescriptor(sd, mcd);
+  }
+
+  /**
+   * Converts a storage descriptor to a db-backed storage descriptor.  Creates a
+   * new db-backed column descriptor object for this SD, unless a matching one already
+   * exists for the given table.
+   * @param sd the storage descriptor to wrap in a db-backed object
+   * @param mt the table to search for existing column descriptors, may be null
+   * @return the storage descriptor db-backed object
+   */
+  private static MStorageDescriptor convertToMStorageDescriptor(StorageDescriptor sd, MTable mt,  ColumnDescriptorSupplier cds)
+      throws MetaException {
+    if (sd == null) {
+      return null;
+    }
+
+    MColumnDescriptor mcd = (mt != null) ? cds.getColumnDescriptor(sd.getCols(), mt) : null;
+    if (mcd == null) {
+      mcd = createNewMColumnDescriptor(sd.getCols());
+    }
     return convertToMStorageDescriptor(sd, mcd);
   }
 
@@ -1959,6 +1980,55 @@ private static MStorageDescriptor convertToMStorageDescriptor(StorageDescriptor
             .getSkewedColValueLocationMaps()), sd.isStoredAsSubDirectories());
   }
 
+  @Override
+  public MColumnDescriptor getColumnDescriptor(List<FieldSchema> cols, MTable mt)
+      throws MetaException {
+    if (cols == null || cols.isEmpty()) {
+      return null;
+    }
+
+    // First check to see if partition and tables column descriptor match
+    // that's the easy and relatively fast-check since does not require
+    // round-tripe to the database
+    List<FieldSchema> tableSchema = mt.getSd() != null && mt.getSd().getCD() != null && mt.getSd()
+        .getCD()
+        .getCols() != null ? convertToFieldSchemas(mt.getSd()
+        .getCD()
+        .getCols()) : null;
+    if (cols.equals(tableSchema)) {
+      return mt.getSd().getCD();
+    }
+
+    String catName = mt.getDatabase().getCatalogName();
+    String dbName = mt.getDatabase().getName();
+    String tblName = mt.getTableName();
+    long tblId = mt.getId();
+    try {
+      return new GetHelper<MColumnDescriptor>(catName, dbName, tblName, true, true) {
+        @Override
+        protected String describeResult() {
+          return "Matching column descriptor";
+        }
+
+        @Override
+        protected MColumnDescriptor getSqlResult(GetHelper<MColumnDescriptor> ctx)
+            throws MetaException {
+          // TODO: Need to make this configurable via property cause it may be too expensive for general use
+          return directSql.getColumnDescriptor(cols, tblId);
+        }
+
+        @Override
+        protected MColumnDescriptor getJdoResult(GetHelper<MColumnDescriptor> ctx) {
+          // Return null basically means allocate a new column descriptor
+          return null;
+        }
+      }.run(false);
+    } catch (NoSuchObjectException e) {
+      return null;
+    }
+  }
+
+
   public static MCreationMetadata convertToMCreationMetadata(CreationMetadata m, RawStore base)
       throws MetaException {
     if (m == null) {
@@ -2066,7 +2136,7 @@ private static SourceTable convertToSourceTable(MMVSource mmvSource, String cata
    * @param mt the parent table object
    * @return the model partition object, and null if the input partition is null.
    */
-  public static MPartition convertToMPart(Partition part, MTable mt)
+  public static MPartition convertToMPart(Partition part, MTable mt, RawStore base)
       throws InvalidObjectException, MetaException {
     // NOTE: we don't set writeId in this method. Write ID is only set after validating the
     //       existing write ID against the caller's valid list.
@@ -2078,20 +2148,13 @@ public static MPartition convertToMPart(Partition part, MTable mt)
           "Partition doesn't have a valid table or database name");
     }
 
-    // If this partition's set of columns is the same as the parent table's,
-    // use the parent table's, so we do not create a duplicate column descriptor,
-    // thereby saving space
-    MStorageDescriptor msd;
-    if (mt.getSd() != null && mt.getSd().getCD() != null &&
-        mt.getSd().getCD().getCols() != null &&
-        part.getSd() != null &&
-        convertToFieldSchemas(mt.getSd().getCD().getCols()).
-        equals(part.getSd().getCols())) {
-      msd = convertToMStorageDescriptor(part.getSd(), mt.getSd().getCD());
-    } else {
-      msd = convertToMStorageDescriptor(part.getSd());
+    ColumnDescriptorSupplier cds = (s, t) -> null;
+    if (base instanceof ColumnDescriptorSupplier) {
+      cds = (ColumnDescriptorSupplier) base;
     }
 
+    MStorageDescriptor msd = convertToMStorageDescriptor(part.getSd(), mt, cds);
+
     return new MPartition(Warehouse.makePartName(convertToFieldSchemas(mt
         .getPartitionKeys()), part.getValues()), mt, part.getValues(), part
         .getCreateTime(), part.getLastAccessTime(),
@@ -6239,7 +6302,7 @@ private MSchemaVersion convertToMSchemaVersion(SchemaVersion schemaVersion) thro
         normalizeIdentifier(schemaVersion.getSchema().getSchemaName())),
         schemaVersion.getVersion(),
         schemaVersion.getCreatedAt(),
-        createNewMColumnDescriptor(convertToMFieldSchemas(schemaVersion.getCols())),
+        createNewMColumnDescriptor(schemaVersion.getCols()),
         schemaVersion.isSetState() ? schemaVersion.getState().getValue() : 0,
         schemaVersion.isSetDescription() ? schemaVersion.getDescription() : null,
         schemaVersion.isSetSchemaText() ? schemaVersion.getSchemaText() : null,

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java

@@ -107,6 +107,7 @@
 import org.apache.hadoop.hive.metastore.client.builder.GetPartitionsArgs;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
 import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
+import org.apache.hadoop.hive.metastore.model.MColumnDescriptor;
 import org.apache.hadoop.hive.metastore.model.MConstraint;
 import org.apache.hadoop.hive.metastore.model.MCreationMetadata;
 import org.apache.hadoop.hive.metastore.model.MDatabase;
@@ -534,6 +535,64 @@ public void addPartitions(List<MPartition> parts, List<List<MPartitionPrivilege>
     directSqlInsertPart.addPartitions(parts, partPrivilegesList, partColPrivilegesList);
   }
 
+  public MColumnDescriptor getColumnDescriptor(List<FieldSchema> cols, long tblId)
+      throws MetaException {
+    if (cols == null || cols.isEmpty()) {
+      return null;
+    }
+    Query<?> query = null;
+    try {
+      // TODO: Is there a more efficient query? Should we add a limit?
+      String findDesciptorsSql = "SELECT DISTINCT \"SDS\".\"CD_ID\" FROM \"SDS\" "
+              + "INNER JOIN \"COLUMNS_V2\" ON \"SDS\".\"CD_ID\" = \"COLUMNS_V2\".\"CD_ID\" "
+              + "WHERE \"SDS\".\"SD_ID\" IN ("
+              + "SELECT \"SD_ID\" FROM \"PARTITIONS\" WHERE \"TBL_ID\" = ?"
+              + ") GROUP BY \"SDS\".\"CD_ID\", \"SDS\".\"SD_ID\" HAVING COUNT(*) = ?";
+      query = pm.newQuery("javax.jdo.query.SQL", findDesciptorsSql);
+      List<Long> candidateIds = executeWithArray(query, new Object[]{tblId, cols.size()}, findDesciptorsSql);
+      if (candidateIds == null || candidateIds.isEmpty()) {
+        return null;
+      }
+
+      for (Long cdId : candidateIds) {
+        if (descriptorHasColumns(cdId, cols)) {
+          return pm.getObjectById(MColumnDescriptor.class, cdId);
+        }
+      }
+    } finally {
+      if (query != null) {
+        query.closeAll();
+      }
+    }
+    return null;
+  }
+
+  private boolean descriptorHasColumns(Long cdId, List<FieldSchema> cols) throws MetaException {
+    String findColumnSql = "SELECT \"COLUMN_NAME\", \"TYPE_NAME\", \"COMMENT\" FROM \"COLUMNS_V2\" "
+        + "WHERE \"CD_ID\" = ? ORDER BY \"INTEGER_IDX\"";
+    Query query = null;
+    try {
+      query = pm.newQuery("javax.jdo.query.SQL", findColumnSql);
+      List<Object[]> rows = executeWithArray(query, new Object[] {cdId}, findColumnSql);
+      if (rows != null && rows.size() == cols.size()) {
+        boolean match = true;
+        for (int i = 0; i < cols.size(); i++) {
+          Object[] row = rows.get(i);
+          FieldSchema col = new FieldSchema(String.valueOf(row[0]), String.valueOf(row[1]), String.valueOf(row[2]));
+          if (!cols.get(i).equals(col)) {
+            match = false;
+            break;
+          }
+        }
+        return match;
+      }
+      return false;
+    } finally {
+      if (query != null) {
+        query.closeAll();
+      }
+    }
+  }
   /**
    * Alter partitions in batch using direct SQL
    * @param table the target table

standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java

@@ -721,7 +721,7 @@ public boolean addPartitions(TableName tableName, List<Partition> parts) throws
         throw new MetaException("Partition does not belong to target table "
             + dbName + "." + tblName + ": " + part);
       }
-      MPartition mpart = convertToMPart(part, table);
+      MPartition mpart = convertToMPart(part, table, baseStore);
       mParts.add(mpart);
       int now = (int) (System.currentTimeMillis() / 1000);
       List<MPartitionPrivilege> mPartPrivileges = new ArrayList<>();
@@ -1901,7 +1901,8 @@ private Partition alterPartitionNoTxn(String catName, String dbname,
     catName = normalizeIdentifier(catName);
     name = normalizeIdentifier(name);
     dbname = normalizeIdentifier(dbname);
-    MPartition newp = convertToMPart(newPart, table);
+    // TODO: Should we perform descriptor reuse in this path?
+    MPartition newp = convertToMPart(newPart, table, baseStore);
     MColumnDescriptor oldCD = null;
     MStorageDescriptor oldSD = oldp.getSd();
     if (oldSD != null) {