diff --git a/ql/src/test/queries/clientpositive/partition_descriptor_reuse.q b/ql/src/test/queries/clientpositive/partition_descriptor_reuse.q new file mode 100644 index 000000000000..6ec6ed6adea8 --- /dev/null +++ b/ql/src/test/queries/clientpositive/partition_descriptor_reuse.q @@ -0,0 +1,19 @@ +-- 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 + +set metastore.add.partition.reuse.existing.column.descriptors=true; + +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; diff --git a/ql/src/test/results/clientpositive/llap/partition_descriptor_reuse.q.out b/ql/src/test/results/clientpositive/llap/partition_descriptor_reuse.q.out new file mode 100644 index 000000000000..f6c6a3d71b08 --- /dev/null +++ b/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 diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java index e955d1e07d67..4cde20bb6e58 100644 --- a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java +++ b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java @@ -17,6 +17,16 @@ */ package org.apache.hadoop.hive.metastore.conf; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.common.ZooKeeperHiveHelper; +import org.apache.hadoop.hive.metastore.utils.StringUtils; +import org.apache.hadoop.security.alias.CredentialProviderFactory; +import org.apache.hive.common.util.SuppressFBWarnings; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import java.io.File; import java.io.IOException; import java.net.URI; @@ -33,17 +43,6 @@ import java.util.regex.Matcher; import java.util.regex.Pattern; -import com.google.common.base.Preconditions; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.hive.common.ZooKeeperHiveHelper; -import org.apache.hadoop.hive.metastore.utils.StringUtils; -import org.apache.hadoop.security.alias.CredentialProviderFactory; -import org.apache.hive.common.util.SuppressFBWarnings; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.google.common.annotations.VisibleForTesting; - /** * A set of definitions of config values used by the Metastore. One of the key aims of this * class is to provide backwards compatibility with existing Hive configuration keys while @@ -312,12 +311,12 @@ public enum ConfVars { ACID_HOUSEKEEPER_SERVICE_INTERVAL("metastore.acid.housekeeper.interval", "hive.metastore.acid.housekeeper.interval", 60, TimeUnit.SECONDS, "Time interval describing how often the acid housekeeper runs."), - COMPACTION_HOUSEKEEPER_SERVICE_INTERVAL("metastore.compaction.housekeeper.interval", - "hive.metastore.compaction.housekeeper.interval", 300, TimeUnit.SECONDS, - "Time interval describing how often the acid compaction housekeeper runs."), ACID_TXN_CLEANER_INTERVAL("metastore.acid.txn.cleaner.interval", "hive.metastore.acid.txn.cleaner.interval", 10, TimeUnit.SECONDS, "Time interval describing how often aborted and committed txns are cleaned."), + ADD_PARTITION_REUSE_EXISTING_COLUMN_DESCRIPTORS("metastore.add.partition.reuse.existing.column.descriptors", + "hive.metastore.add.partition.reuse.existing.column.descriptors", false, + "Add partition reuse existing column descriptors to avoid metadata bloat on schema evolution."), ADDED_JARS("metastore.added.jars.path", "hive.added.jars.path", "", "This an internal parameter."), AGGREGATE_STATS_CACHE_CLEAN_UNTIL("metastore.aggregate.stats.cache.clean.until", @@ -432,6 +431,9 @@ public enum ConfVars { TimeUnit.SECONDS, "MetaStore Client socket timeout in seconds"), CLIENT_CONNECTION_TIMEOUT("metastore.client.connection.timeout", "hive.metastore.client.connection.timeout", 600, TimeUnit.SECONDS, "MetaStore Client connection timeout in seconds"), + COMPACTION_HOUSEKEEPER_SERVICE_INTERVAL("metastore.compaction.housekeeper.interval", + "hive.metastore.compaction.housekeeper.interval", 300, TimeUnit.SECONDS, + "Time interval describing how often the acid compaction housekeeper runs."), COMPACTOR_HISTORY_RETENTION_DID_NOT_INITIATE("metastore.compactor.history.retention.did.not.initiate", "hive.compactor.history.retention.did.not.initiate", 2, new RangeValidator(0, 100), "Determines how many compaction records in state " + diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ColumnDescriptorSupplier.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ColumnDescriptorSupplier.java new file mode 100644 index 000000000000..e73ae0ca40e5 --- /dev/null +++ b/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 cols, MTable mt) throws MetaException; +} diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java index 17ad48698ffb..0876ffed2234 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/ObjectStore.java @@ -18,55 +18,19 @@ package org.apache.hadoop.hive.metastore; -import static org.apache.hadoop.hive.metastore.Batchable.NO_BATCHING; -import static org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars.COMPACTOR_USE_CUSTOM_POOL; -import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog; -import static org.apache.hadoop.hive.metastore.utils.StringUtils.normalizeIdentifier; -import static org.apache.hadoop.hive.metastore.utils.StringUtils.normalizeIdentifiers; - -import java.io.IOException; -import java.net.InetAddress; -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.SQLIntegrityConstraintViolationException; -import java.sql.Statement; -import java.time.Instant; -import java.time.LocalDateTime; -import java.time.ZoneId; -import java.time.ZonedDateTime; -import java.time.format.DateTimeFormatter; -import java.util.ArrayList; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Iterator; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.Map.Entry; -import java.util.Optional; -import java.util.Set; -import java.util.TreeMap; -import java.util.UUID; -import java.util.concurrent.ThreadLocalRandom; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Consumer; -import java.util.regex.Pattern; -import java.util.stream.Collectors; - -import javax.jdo.JDODataStoreException; -import javax.jdo.JDOException; -import javax.jdo.JDOObjectNotFoundException; -import javax.jdo.PersistenceManager; -import javax.jdo.Query; -import javax.jdo.Transaction; -import javax.jdo.datastore.JDOConnection; -import javax.jdo.identity.IntIdentity; - +import com.codahale.metrics.Counter; +import com.codahale.metrics.MetricRegistry; +import com.cronutils.model.CronType; +import com.cronutils.model.definition.CronDefinition; +import com.cronutils.model.definition.CronDefinitionBuilder; +import com.cronutils.model.time.ExecutionTime; +import com.cronutils.parser.CronParser; +import com.google.common.annotations.VisibleForTesting; +import com.google.common.base.Preconditions; +import com.google.common.base.Strings; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.common.collect.Sets; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.ArrayUtils; import org.apache.commons.lang3.StringUtils; @@ -81,9 +45,7 @@ import org.apache.hadoop.hive.common.TableName; import org.apache.hadoop.hive.common.ValidReaderWriteIdList; import org.apache.hadoop.hive.common.ValidWriteIdList; -import org.apache.hadoop.hive.metastore.directsql.DirectSqlDeleteStats; -import org.apache.hadoop.hive.metastore.directsql.MetaStoreDirectSql; -import org.apache.hadoop.hive.metastore.directsql.MetaStoreDirectSql.SqlFilterForPushdown; +import org.apache.hadoop.hive.metastore.api.AddPackageRequest; import org.apache.hadoop.hive.metastore.api.AggrStats; import org.apache.hadoop.hive.metastore.api.AllTableConstraintsRequest; import org.apache.hadoop.hive.metastore.api.AlreadyExistsException; @@ -93,12 +55,11 @@ import org.apache.hadoop.hive.metastore.api.ColumnStatisticsDesc; import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj; import org.apache.hadoop.hive.metastore.api.CreationMetadata; +import org.apache.hadoop.hive.metastore.api.DataConnector; import org.apache.hadoop.hive.metastore.api.Database; -import org.apache.hadoop.hive.metastore.api.AddPackageRequest; +import org.apache.hadoop.hive.metastore.api.DatabaseType; import org.apache.hadoop.hive.metastore.api.DefaultConstraintsRequest; import org.apache.hadoop.hive.metastore.api.DropPackageRequest; -import org.apache.hadoop.hive.metastore.api.DatabaseType; -import org.apache.hadoop.hive.metastore.api.DataConnector; import org.apache.hadoop.hive.metastore.api.FieldSchema; import org.apache.hadoop.hive.metastore.api.ForeignKeysRequest; import org.apache.hadoop.hive.metastore.api.Function; @@ -106,6 +67,7 @@ import org.apache.hadoop.hive.metastore.api.GetPackageRequest; import org.apache.hadoop.hive.metastore.api.GetPartitionsFilterSpec; import org.apache.hadoop.hive.metastore.api.GetProjectionsSpec; +import org.apache.hadoop.hive.metastore.api.GetReplicationMetricsRequest; import org.apache.hadoop.hive.metastore.api.ISchema; import org.apache.hadoop.hive.metastore.api.ISchemaName; import org.apache.hadoop.hive.metastore.api.InvalidInputException; @@ -124,12 +86,11 @@ import org.apache.hadoop.hive.metastore.api.PrincipalType; import org.apache.hadoop.hive.metastore.api.PrivilegeGrantInfo; import org.apache.hadoop.hive.metastore.api.QueryState; +import org.apache.hadoop.hive.metastore.api.ReplicationMetricList; +import org.apache.hadoop.hive.metastore.api.ReplicationMetrics; import org.apache.hadoop.hive.metastore.api.ResourceType; import org.apache.hadoop.hive.metastore.api.ResourceUri; import org.apache.hadoop.hive.metastore.api.RuntimeStat; -import org.apache.hadoop.hive.metastore.api.ReplicationMetricList; -import org.apache.hadoop.hive.metastore.api.GetReplicationMetricsRequest; -import org.apache.hadoop.hive.metastore.api.ReplicationMetrics; import org.apache.hadoop.hive.metastore.api.SQLAllTableConstraints; import org.apache.hadoop.hive.metastore.api.SQLCheckConstraint; import org.apache.hadoop.hive.metastore.api.SQLDefaultConstraint; @@ -173,6 +134,13 @@ import org.apache.hadoop.hive.metastore.conf.MetastoreConf; import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars; import org.apache.hadoop.hive.metastore.directsql.DirectSqlAggrStats; +import org.apache.hadoop.hive.metastore.directsql.DirectSqlDeleteStats; +import org.apache.hadoop.hive.metastore.directsql.MetaStoreDirectSql; +import org.apache.hadoop.hive.metastore.directsql.MetaStoreDirectSql.SqlFilterForPushdown; +import org.apache.hadoop.hive.metastore.metastore.MetaDescriptor; +import org.apache.hadoop.hive.metastore.metastore.PersistenceManagerProxy; +import org.apache.hadoop.hive.metastore.metastore.RawStoreAware; +import org.apache.hadoop.hive.metastore.metastore.TransactionHandler; import org.apache.hadoop.hive.metastore.metastore.iface.PrivilegeStore; import org.apache.hadoop.hive.metastore.metrics.Metrics; import org.apache.hadoop.hive.metastore.metrics.MetricsConstants; @@ -181,8 +149,8 @@ import org.apache.hadoop.hive.metastore.model.MConstraint; import org.apache.hadoop.hive.metastore.model.MCreationMetadata; import org.apache.hadoop.hive.metastore.model.MDBPrivilege; -import org.apache.hadoop.hive.metastore.model.MDataConnector; import org.apache.hadoop.hive.metastore.model.MDCPrivilege; +import org.apache.hadoop.hive.metastore.model.MDataConnector; import org.apache.hadoop.hive.metastore.model.MDatabase; import org.apache.hadoop.hive.metastore.model.MDelegationToken; import org.apache.hadoop.hive.metastore.model.MFieldSchema; @@ -196,6 +164,7 @@ import org.apache.hadoop.hive.metastore.model.MPartition; import org.apache.hadoop.hive.metastore.model.MPartitionColumnStatistics; import org.apache.hadoop.hive.metastore.model.MPartitionEvent; +import org.apache.hadoop.hive.metastore.model.MReplicationMetrics; import org.apache.hadoop.hive.metastore.model.MResourceUri; import org.apache.hadoop.hive.metastore.model.MRuntimeStat; import org.apache.hadoop.hive.metastore.model.MScheduledExecution; @@ -216,13 +185,8 @@ import org.apache.hadoop.hive.metastore.model.MWMResourcePlan; import org.apache.hadoop.hive.metastore.model.MWMResourcePlan.Status; import org.apache.hadoop.hive.metastore.model.MWMTrigger; -import org.apache.hadoop.hive.metastore.model.MReplicationMetrics; import org.apache.hadoop.hive.metastore.properties.CachingPropertyStore; import org.apache.hadoop.hive.metastore.properties.PropertyStore; -import org.apache.hadoop.hive.metastore.metastore.PersistenceManagerProxy; -import org.apache.hadoop.hive.metastore.metastore.RawStoreAware; -import org.apache.hadoop.hive.metastore.metastore.MetaDescriptor; -import org.apache.hadoop.hive.metastore.metastore.TransactionHandler; import org.apache.hadoop.hive.metastore.tools.SQLGenerator; import org.apache.hadoop.hive.metastore.txn.TxnUtils; import org.apache.hadoop.hive.metastore.utils.JavaUtils; @@ -236,20 +200,54 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.codahale.metrics.Counter; -import com.codahale.metrics.MetricRegistry; -import com.cronutils.model.CronType; -import com.cronutils.model.definition.CronDefinition; -import com.cronutils.model.definition.CronDefinitionBuilder; -import com.cronutils.model.time.ExecutionTime; -import com.cronutils.parser.CronParser; -import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Preconditions; -import com.google.common.base.Strings; -import com.google.common.collect.Lists; -import com.google.common.collect.Maps; -import com.google.common.collect.Sets; +import javax.jdo.JDODataStoreException; +import javax.jdo.JDOException; +import javax.jdo.JDOObjectNotFoundException; +import javax.jdo.PersistenceManager; +import javax.jdo.Query; +import javax.jdo.Transaction; +import javax.jdo.datastore.JDOConnection; +import javax.jdo.identity.IntIdentity; +import java.io.IOException; +import java.net.InetAddress; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.SQLIntegrityConstraintViolationException; +import java.sql.Statement; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.time.ZonedDateTime; +import java.time.format.DateTimeFormatter; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; +import java.util.UUID; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import java.util.regex.Pattern; +import java.util.stream.Collectors; +import static org.apache.hadoop.hive.metastore.Batchable.NO_BATCHING; +import static org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars.ADD_PARTITION_REUSE_EXISTING_COLUMN_DESCRIPTORS; +import static org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars.COMPACTOR_USE_CUSTOM_POOL; +import static org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog; +import static org.apache.hadoop.hive.metastore.utils.StringUtils.normalizeIdentifier; +import static org.apache.hadoop.hive.metastore.utils.StringUtils.normalizeIdentifiers; /** * This class is the interface between the application logic and the database @@ -258,7 +256,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 +1819,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 cols) { + private static MColumnDescriptor createNewMColumnDescriptor(List cols) { if (cols == null) { return null; } - return new MColumnDescriptor(cols); + return new MColumnDescriptor(convertToMFieldSchemas(cols)); } private static StorageDescriptor convertToStorageDescriptor( @@ -1929,7 +1927,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 +1978,59 @@ private static MStorageDescriptor convertToMStorageDescriptor(StorageDescriptor .getSkewedColValueLocationMaps()), sd.isStoredAsSubDirectories()); } + @Override + @VisibleForTesting + public MColumnDescriptor getColumnDescriptor(List 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 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(catName, dbName, tblName, true, true) { + @Override + protected String describeResult() { + return "Matching column descriptor"; + } + + @Override + protected MColumnDescriptor getSqlResult(GetHelper ctx) + throws MetaException { + if (MetastoreConf.getBoolVar(getConf(), ADD_PARTITION_REUSE_EXISTING_COLUMN_DESCRIPTORS)) { + return directSql.getColumnDescriptor(cols, tblId); + } + // Return null basically means allocate a new column descriptor + return null; + } + + @Override + protected MColumnDescriptor getJdoResult(GetHelper 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 +2138,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 +2150,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 +6304,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, diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java index 713f3ff4e338..b995661a6825 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java @@ -18,48 +18,10 @@ package org.apache.hadoop.hive.metastore.directsql; -import static org.apache.commons.lang3.StringUtils.join; -import static org.apache.commons.lang3.StringUtils.normalizeSpace; -import static org.apache.hadoop.hive.metastore.ColumnType.BIGINT_TYPE_NAME; -import static org.apache.hadoop.hive.metastore.ColumnType.CHAR_TYPE_NAME; -import static org.apache.hadoop.hive.metastore.ColumnType.DATE_TYPE_NAME; -import static org.apache.hadoop.hive.metastore.ColumnType.INT_TYPE_NAME; -import static org.apache.hadoop.hive.metastore.ColumnType.SMALLINT_TYPE_NAME; -import static org.apache.hadoop.hive.metastore.ColumnType.STRING_TYPE_NAME; -import static org.apache.hadoop.hive.metastore.ColumnType.TIMESTAMP_TYPE_NAME; -import static org.apache.hadoop.hive.metastore.ColumnType.TINYINT_TYPE_NAME; -import static org.apache.hadoop.hive.metastore.ColumnType.VARCHAR_TYPE_NAME; -import static org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.getFullyQualifiedName; -import static org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.makeParams; -import static org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.prepareParams; - -import java.sql.Connection; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Iterator; -import java.util.LinkedList; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.Set; -import java.util.TreeMap; -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.atomic.AtomicReference; -import java.util.stream.Collectors; - -import javax.jdo.PersistenceManager; -import javax.jdo.Query; -import javax.jdo.Transaction; -import javax.jdo.datastore.JDOConnection; - import com.google.common.collect.ImmutableList; +import com.google.common.collect.ImmutableMap; import com.google.common.collect.ImmutableSet; +import com.google.common.collect.Lists; import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hive.common.StatsSetupConst; @@ -107,6 +69,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; @@ -135,8 +98,44 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.collect.ImmutableMap; -import com.google.common.collect.Lists; +import javax.jdo.PersistenceManager; +import javax.jdo.Query; +import javax.jdo.Transaction; +import javax.jdo.datastore.JDOConnection; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.TreeMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; + +import static org.apache.commons.lang3.StringUtils.join; +import static org.apache.commons.lang3.StringUtils.normalizeSpace; +import static org.apache.hadoop.hive.metastore.ColumnType.BIGINT_TYPE_NAME; +import static org.apache.hadoop.hive.metastore.ColumnType.CHAR_TYPE_NAME; +import static org.apache.hadoop.hive.metastore.ColumnType.DATE_TYPE_NAME; +import static org.apache.hadoop.hive.metastore.ColumnType.INT_TYPE_NAME; +import static org.apache.hadoop.hive.metastore.ColumnType.SMALLINT_TYPE_NAME; +import static org.apache.hadoop.hive.metastore.ColumnType.STRING_TYPE_NAME; +import static org.apache.hadoop.hive.metastore.ColumnType.TIMESTAMP_TYPE_NAME; +import static org.apache.hadoop.hive.metastore.ColumnType.TINYINT_TYPE_NAME; +import static org.apache.hadoop.hive.metastore.ColumnType.VARCHAR_TYPE_NAME; +import static org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.getFullyQualifiedName; +import static org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.makeParams; +import static org.apache.hadoop.hive.metastore.directsql.MetastoreDirectSqlUtils.prepareParams; /** * This class contains the optimizations for MetaStore that rely on direct SQL access to @@ -534,6 +533,68 @@ public void addPartitions(List parts, List directSqlInsertPart.addPartitions(parts, partPrivilegesList, partColPrivilegesList); } + public MColumnDescriptor getColumnDescriptor(List 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 + PARTITIONS + INNER JOIN SDS ON SDS.SD_ID = PARTITIONS.SD_ID + INNER JOIN COLUMNS_V2 ON SDS.CD_ID = COLUMNS_V2.CD_ID + WHERE TBL_ID = ? + GROUP BY SDS.CD_ID, SDS.SD_ID + HAVING COUNT(*) = ?"""; + query = pm.newQuery("javax.jdo.query.SQL", findDesciptorsSql); + List 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 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 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 diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java index c03e1b8e0016..c3c25af84250 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java +++ b/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 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 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) { diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHMSColumnDescriptorReuse.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHMSColumnDescriptorReuse.java new file mode 100644 index 000000000000..e8c250f951b8 --- /dev/null +++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHMSColumnDescriptorReuse.java @@ -0,0 +1,162 @@ +/* + * 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.conf.Configuration; +import org.apache.hadoop.hive.metastore.annotation.MetastoreUnitTest; +import org.apache.hadoop.hive.metastore.api.Database; +import org.apache.hadoop.hive.metastore.api.FieldSchema; +import org.apache.hadoop.hive.metastore.api.InvalidInputException; +import org.apache.hadoop.hive.metastore.api.InvalidObjectException; +import org.apache.hadoop.hive.metastore.api.MetaException; +import org.apache.hadoop.hive.metastore.api.NoSuchObjectException; +import org.apache.hadoop.hive.metastore.api.Partition; +import org.apache.hadoop.hive.metastore.api.SerDeInfo; +import org.apache.hadoop.hive.metastore.api.StorageDescriptor; +import org.apache.hadoop.hive.metastore.api.Table; +import org.apache.hadoop.hive.metastore.client.builder.DatabaseBuilder; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.hadoop.hive.metastore.dbinstall.rules.DatabaseRule; +import org.apache.hadoop.hive.metastore.dbinstall.rules.Postgres; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; + +import static org.apache.hadoop.hive.metastore.Warehouse.DEFAULT_CATALOG_NAME; +import static org.junit.Assert.assertEquals; + +@Category(MetastoreUnitTest.class) +public class TestHMSColumnDescriptorReuse { + private ObjectStore objectStore = null; + // Use Postgres instead of Derby for facilitate debugging and looking into the database + // In the final version we should rather use Derby + // TODO: Beore merging we should test with all supported DBMS + private static final DatabaseRule DB = new Postgres(); + + @Before + public void setUp() throws Exception { + DB.before(); + DB.install(); + Configuration conf = MetastoreConf.newMetastoreConf(); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CONNECT_URL_KEY, DB.getJdbcUrl()); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CONNECTION_DRIVER, DB.getJdbcDriver()); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CONNECTION_USER_NAME, DB.getHiveUser()); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.PWD, DB.getHivePassword()); + MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.AUTO_CREATE_ALL, false); + + MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.HIVE_IN_TEST, true); + MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.ADD_PARTITION_REUSE_EXISTING_COLUMN_DESCRIPTORS, true); + + MetaStoreTestUtils.setConfForStandloneMode(conf); + + objectStore = new ObjectStore(); + objectStore.setConf(conf); + HMSHandler.createDefaultCatalog(objectStore, new Warehouse(conf)); + Database db = new DatabaseBuilder() + .setName("default") + .setDescription("description") + .setLocation("locationurl") + .build(conf); + objectStore.createDatabase(db); + } + + @After + public void tearDown() throws Exception { + DB.after(); + } + + @Test + public void testAddPartitionAlterAddPartition() throws MetaException, InvalidObjectException, InvalidInputException, NoSuchObjectException { + FieldSchema id = new FieldSchema("id", ColumnType.STRING_TYPE_NAME, ""); + FieldSchema fname = new FieldSchema("fname", ColumnType.STRING_TYPE_NAME, ""); + FieldSchema country = new FieldSchema("country", ColumnType.STRING_TYPE_NAME, ""); + + Table tbl1 = newTable("person", Arrays.asList(id, fname), Collections.singletonList(country)); + objectStore.createTable(tbl1); + objectStore.addPartition(newPart(tbl1, "US")); + objectStore.addPartition(newPart(tbl1, "Greece")); + FieldSchema lname = new FieldSchema("lname", ColumnType.STRING_TYPE_NAME, ""); + Table tbl2 = newTable("person", Arrays.asList(id, fname, lname), Collections.singletonList(country)); + objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(), tbl1.getTableName(), tbl2, null); + objectStore.addPartition(newPart(tbl2, "Italy")); + // Mimics replication scenario where we are adding partitions to the "same" table but with a different schema. + // The tbl1 is using the old storage descriptor so "Germany" and "Belgium" partitions will have the old schema + // And will lead to "duplicate" entries in "CDS" and "COLUMNS_V2" tables. + objectStore.addPartition(newPart(tbl1, "Germany")); + objectStore.addPartition(newPart(tbl1, "Belgium")); + // On the other hand the addition of a partition to the table with the new schema is successfully using the + // existing storage/column descriptors + objectStore.addPartition(newPart(tbl2, "England")); + // This assertion could be more elaborate and not just a simple count + assertEquals(2, countColumnDescriptors()); + } + + + private int countColumnDescriptors() { + try(Connection c = DriverManager.getConnection(DB.getJdbcUrl(), DB.getHiveUser(), DB.getHivePassword())){ + try(ResultSet rs = c.prepareStatement("SELECT COUNT(*) FROM \"CDS\"").executeQuery()) { + rs.next(); + return rs.getInt(1); + } + } catch (SQLException e) { + throw new RuntimeException(e); + } + } + + private static Table newTable(String name, List columns, List partCols ) { + int timeSec = (int) System.currentTimeMillis() / 1000; + StorageDescriptor sd = new StorageDescriptor(columns, + "/fake/location/person", + "org.apache.hadoop.mapred.TextInputFormat", + "org.apache.hadoop.mapred.MapFileOutputFormat", + false, + 0, + new SerDeInfo("SerDeName", "serializationLib", null), + null, + null, + null); + HashMap tableParams = new HashMap<>(); + tableParams.put("EXTERNAL", "false"); + return + new Table(name, "default", "owner", timeSec, timeSec, 3, sd, partCols, + tableParams, null, null, "MANAGED_TABLE"); + } + + private static Partition newPart(Table tbl, String value) { + int timeSec = (int) System.currentTimeMillis() / 1000; + HashMap partitionParams = new HashMap<>(); + partitionParams.put("PARTITION_LEVEL_PRIVILEGE", "true"); + StorageDescriptor psd = tbl.getSd().deepCopy(); + psd.setLocation(psd.getLocation() + "/" + value); + Partition p = new Partition(Collections.singletonList(value), tbl.getDbName(), tbl.getTableName(), timeSec, timeSec, psd, partitionParams); + p.setCatName(DEFAULT_CATALOG_NAME); + return p; + } + +} +