Copilot commented on code in PR #6577:
URL: https://github.com/apache/hive/pull/6577#discussion_r3548499099
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetastoreDirectSqlUtils.java:
##########
@@ -582,6 +582,12 @@ public static Boolean extractSqlBoolean(Object value)
throws MetaException {
public static String extractSqlString(Object value) {
if (value == null) return null;
+
+ // workaround for DataNucleus's Oracle empty string workaround (it puts an
ASCII value 1 instead of empty string
Review Comment:
Comment typo: the parenthesis is not closed, and the sentence is a bit
unclear. Please fix the wording to avoid leaving malformed comments in the
codebase.
##########
standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHMSColumnDescriptorReuse.java:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.InvalidObjectException;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+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.Derby;
+import org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil;
+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.ResultSet;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+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;
+ Configuration conf;
+
+ // Modify to try out with different databases.
+ // Keep it on Derby once you commit your change. It makes the test execution
faster.
+ private final DatabaseRule DB = new Derby(true);
+
+ @Before
+ public void setUp() throws Exception {
+ DB.before();
+ DB.install();
+
+ 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.PARTITION_REUSE_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() {
+ DB.after();
+ }
+
+ @Test
+ public void testReuseAfterSimpleSchemaEvolution() throws Exception {
+ 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(Arrays.asList(id, fname),
Collections.singletonList(country));
+ objectStore.createTable(tbl1);
+ objectStore.addPartition(newPart(tbl1, "US"));
+
+ FieldSchema lname = new FieldSchema("lname", ColumnType.STRING_TYPE_NAME,
"");
+ Table tbl2 = newTable(Arrays.asList(id, fname, lname),
Collections.singletonList(country));
+
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl2, null);
+ objectStore.addPartition(newPart(tbl2, "Italy"));
+
+ objectStore.addPartition(newPart(tbl2, "Hungary"));
+ objectStore.addPartition(newPart(tbl1, "Ukraine"));
+
+ assertEquals(2, countColumnDescriptors());
+ }
+
+ @Test
+ public void testAddPartitionAlterAddPartition() throws MetaException,
InvalidObjectException {
+ 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(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(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"));
+
+ assertEquals(2, countColumnDescriptors());
+ }
+
+ @Test
+ public void testNoReusableColumnDescriptors() throws MetaException,
InvalidObjectException {
+ 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(Arrays.asList(id, fname),
Collections.singletonList(country));
+ objectStore.createTable(tbl1);
+ objectStore.addPartition(newPart(tbl1, "US"));
+
+ FieldSchema lname = new FieldSchema("lname", ColumnType.STRING_TYPE_NAME,
"");
+ Table tbl2 = newTable(Arrays.asList(id, fname, lname),
Collections.singletonList(country));
+
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl2, null);
+ objectStore.addPartition(newPart(tbl2, "Italy"));
+
+ FieldSchema address = new FieldSchema("address",
ColumnType.STRING_TYPE_NAME, "");
+ Table tbl3 = newTable(Arrays.asList(id, fname, lname, address),
Collections.singletonList(country));
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl3, null);
+ objectStore.addPartition(newPart(tbl3, "Hungary"));
+
+ assertEquals(3, countColumnDescriptors());
+ }
+
+ private int countColumnDescriptors() {
+ try(Connection c = TestTxnDbUtil.getConnection(conf)){
+ try(ResultSet rs = c.prepareStatement("SELECT COUNT(*) FROM
\"CDS\"").executeQuery()) {
+ rs.next();
+ return rs.getInt(1);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Table newTable(List<FieldSchema> columns, List<FieldSchema>
partCols) {
+ int timeSec = (int) System.currentTimeMillis() / 1000;
Review Comment:
The timestamp computation casts before dividing, which can overflow
(currentTimeMillis exceeds Integer.MAX_VALUE) and produce a negative/incorrect
time. Cast after dividing instead.
##########
standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/TestHMSColumnDescriptorReuse.java:
##########
@@ -0,0 +1,221 @@
+/*
+ * 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.InvalidObjectException;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+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.Derby;
+import org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil;
+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.ResultSet;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+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;
+ Configuration conf;
+
+ // Modify to try out with different databases.
+ // Keep it on Derby once you commit your change. It makes the test execution
faster.
+ private final DatabaseRule DB = new Derby(true);
+
+ @Before
+ public void setUp() throws Exception {
+ DB.before();
+ DB.install();
+
+ 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.PARTITION_REUSE_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() {
+ DB.after();
+ }
+
+ @Test
+ public void testReuseAfterSimpleSchemaEvolution() throws Exception {
+ 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(Arrays.asList(id, fname),
Collections.singletonList(country));
+ objectStore.createTable(tbl1);
+ objectStore.addPartition(newPart(tbl1, "US"));
+
+ FieldSchema lname = new FieldSchema("lname", ColumnType.STRING_TYPE_NAME,
"");
+ Table tbl2 = newTable(Arrays.asList(id, fname, lname),
Collections.singletonList(country));
+
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl2, null);
+ objectStore.addPartition(newPart(tbl2, "Italy"));
+
+ objectStore.addPartition(newPart(tbl2, "Hungary"));
+ objectStore.addPartition(newPart(tbl1, "Ukraine"));
+
+ assertEquals(2, countColumnDescriptors());
+ }
+
+ @Test
+ public void testAddPartitionAlterAddPartition() throws MetaException,
InvalidObjectException {
+ 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(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(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"));
+
+ assertEquals(2, countColumnDescriptors());
+ }
+
+ @Test
+ public void testNoReusableColumnDescriptors() throws MetaException,
InvalidObjectException {
+ 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(Arrays.asList(id, fname),
Collections.singletonList(country));
+ objectStore.createTable(tbl1);
+ objectStore.addPartition(newPart(tbl1, "US"));
+
+ FieldSchema lname = new FieldSchema("lname", ColumnType.STRING_TYPE_NAME,
"");
+ Table tbl2 = newTable(Arrays.asList(id, fname, lname),
Collections.singletonList(country));
+
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl2, null);
+ objectStore.addPartition(newPart(tbl2, "Italy"));
+
+ FieldSchema address = new FieldSchema("address",
ColumnType.STRING_TYPE_NAME, "");
+ Table tbl3 = newTable(Arrays.asList(id, fname, lname, address),
Collections.singletonList(country));
+ objectStore.alterTable(DEFAULT_CATALOG_NAME, tbl1.getDbName(),
tbl1.getTableName(), tbl3, null);
+ objectStore.addPartition(newPart(tbl3, "Hungary"));
+
+ assertEquals(3, countColumnDescriptors());
+ }
+
+ private int countColumnDescriptors() {
+ try(Connection c = TestTxnDbUtil.getConnection(conf)){
+ try(ResultSet rs = c.prepareStatement("SELECT COUNT(*) FROM
\"CDS\"").executeQuery()) {
+ rs.next();
+ return rs.getInt(1);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static Table newTable(List<FieldSchema> columns, List<FieldSchema>
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);
+
+ Map<String, String> tableParams = new HashMap<>();
+ tableParams.put("EXTERNAL", "false");
+
+ return
+ new Table("person", "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;
Review Comment:
Same overflow/precedence issue here: cast happens before dividing, which can
corrupt the timestamp value.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java:
##########
@@ -534,6 +536,150 @@ public void addPartitions(List<MPartition> parts,
List<List<MPartitionPrivilege>
directSqlInsertPart.addPartitions(parts, partPrivilegesList,
partColPrivilegesList);
}
+ /**
+ * Gets a suitable column descriptor for an existing table, based on the
latest partitions
+ *
+ * @param cols column list of a partition
+ * @param tblId table id
+ * @return an existing column descriptor which columns matches with @cols.
Null if there is no match.
+ * @throws MetaException
+ */
+ public MColumnDescriptor getColumnDescriptor(List<FieldSchema> cols, long
tblId)
+ throws MetaException {
+
+ if (cols == null || cols.isEmpty()) {
+ return null;
+ }
+
+ // Please note! In case you modify any of those methods, run
TestHMSColumnDescriptorReuse with all the supported databases
+ List<Long> cdCandidates = findTheLatestColumnDescriptors(tblId);
+
+ cdCandidates = filterCandidatesByColumnCount(cdCandidates, cols.size());
+
+ if (cdCandidates == null || cdCandidates.isEmpty()) {
+ return null;
+ }
+
+ Long matchedColumnDescriptorId =
matchColumnDescriptorWithActualColumns(cdCandidates, cols);
+
+ if (matchedColumnDescriptorId != null) {
+ return pm.getObjectById(MColumnDescriptor.class,
matchedColumnDescriptorId);
+ }
+
+ return null;
+ }
+
+ private List<Long> findTheLatestColumnDescriptors(long tblId) throws
MetaException {
+ final int limit = 20;
+
+ String findLatestDescriptorsSql =
+ """
+ SELECT s."CD_ID"
+ FROM "PARTITIONS" p
+ JOIN "SDS" s ON s."SD_ID" = p."SD_ID"
+ WHERE p."TBL_ID" = ?
+ GROUP BY s."CD_ID"
+ ORDER BY MAX(p."PART_ID") DESC
+ """;
+
+ try (QueryWrapper query = new
QueryWrapper(pm.newQuery("javax.jdo.query.SQL", findLatestDescriptorsSql))) {
+
+ List<Object> sqlResult = executeWithArray(
+ query.getInnerQuery(), new Object[] {tblId},
findLatestDescriptorsSql, limit);
+
+ if (sqlResult == null || sqlResult.isEmpty()) {
+ return null;
+ }
+
+ List<Long> latestColumnDescriptorIds = new ArrayList<>();
+ for (Object cdId : sqlResult) {
+
latestColumnDescriptorIds.add(MetastoreDirectSqlUtils.extractSqlLong(cdId));
+ }
+ return latestColumnDescriptorIds;
+ }
+ }
+
+ private List<Long> filterCandidatesByColumnCount(List<Long> cdCandidates,
int size) throws MetaException {
+ if (cdCandidates == null || cdCandidates.isEmpty()) {
+ return null;
+ }
+
+ String placeholders = cdCandidates.stream()
+ .map(c -> "?")
+ .collect(Collectors.joining(", "));
+
+ String candidatesWithProperColumnCountSql = String.format(
+ """
+ SELECT c."CD_ID"
+ FROM "COLUMNS_V2" c
+ WHERE c."CD_ID" IN (%s)
+ GROUP BY c."CD_ID"
+ HAVING COUNT(*) = ?
+ """, placeholders);
+
+ try (
+ QueryWrapper query =
+ new QueryWrapper(pm.newQuery("javax.jdo.query.SQL",
candidatesWithProperColumnCountSql))
+ ){
+
+ Object[] params = new Object[cdCandidates.size() + 1];
+ for (int i = 0; i < cdCandidates.size(); i++) {
+ params[i] = cdCandidates.get(i);
+ }
+ params[cdCandidates.size()] = size;
+
+ List<Object> result = executeWithArray(query.getInnerQuery(), params,
candidatesWithProperColumnCountSql);
+
+ if (result == null || result.isEmpty()) {
+ return null;
+ }
+
+ List<Long> candidateIds = new ArrayList<>();
+ for (Object cdId : result) {
+ candidateIds.add(MetastoreDirectSqlUtils.extractSqlLong(cdId));
+ }
+
+ return candidateIds;
+
+ }
+ }
+
+ private Long matchColumnDescriptorWithActualColumns(List<Long> cdCandidates,
List<FieldSchema> cols)
+ throws MetaException {
+ if (cdCandidates == null || cdCandidates.isEmpty()) {
+ return null;
+ }
+
+ String findColumnSql = "SELECT \"COLUMN_NAME\", \"TYPE_NAME\", \"COMMENT\"
FROM \"COLUMNS_V2\" "
+ + "WHERE \"CD_ID\" = ? ORDER BY \"INTEGER_IDX\"";
+
+ for (Long candidate: cdCandidates) {
+ try (QueryWrapper query = new
QueryWrapper(pm.newQuery("javax.jdo.query.SQL", findColumnSql))) {
+
+ List<Object[]> rows = executeWithArray(query.getInnerQuery(), new
Object[] {candidate}, findColumnSql);
+ if (rows != null && rows.size() == cols.size()) {
+ int i = 0;
+ for (; i < cols.size(); i++) {
+ Object[] row = rows.get(i);
+ FieldSchema col = new FieldSchema(
+ extractSqlString(row[0]),
+ extractSqlClob(row[1]),
+ extractSqlString(row[2]));
+ if (!cols.get(i).equals(col)) {
+ break;
+ }
Review Comment:
The PR description says to match candidates by column names and types, but
this comparison uses FieldSchema.equals, which also compares the COMMENT field.
This can prevent reuse when comments differ, reintroducing metadata bloat for
schema-evolved partitions. Compare only name+type (and optionally normalize
case/whitespace if needed) to match the intended algorithm.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java:
##########
@@ -3331,26 +3332,85 @@ private 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());
- }
+ MStorageDescriptor msd = convertToMStorageDescriptor(part.getSd(), mt);
return new MPartition(Warehouse.makePartName(convertToFieldSchemas(mt
.getPartitionKeys()), part.getValues()), mt, part.getValues(), part
.getCreateTime(), part.getLastAccessTime(),
msd, part.getParameters());
}
+ /**
+ * 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 MStorageDescriptor convertToMStorageDescriptor(StorageDescriptor sd,
MTable mt)
+ throws MetaException {
+ if (sd == null) {
+ return null;
+ }
+
+ MColumnDescriptor mcd = (mt != null) ? getColumnDescriptor(sd.getCols(),
mt) : null;
+ if (mcd == null) {
+ mcd = createNewMColumnDescriptor(convertToMFieldSchemas(sd.getCols()));
+ }
+ return convertToMStorageDescriptor(sd, mcd);
+ }
+
+ public MColumnDescriptor getColumnDescriptor(List<FieldSchema> cols, MTable
mt)
Review Comment:
This helper is only used inside TableStoreImpl; making it public
unnecessarily expands the class API surface and invites external callers to
depend on internal behavior.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/directsql/MetaStoreDirectSql.java:
##########
@@ -534,6 +536,150 @@ public void addPartitions(List<MPartition> parts,
List<List<MPartitionPrivilege>
directSqlInsertPart.addPartitions(parts, partPrivilegesList,
partColPrivilegesList);
}
+ /**
+ * Gets a suitable column descriptor for an existing table, based on the
latest partitions
+ *
+ * @param cols column list of a partition
+ * @param tblId table id
+ * @return an existing column descriptor which columns matches with @cols.
Null if there is no match.
+ * @throws MetaException
+ */
+ public MColumnDescriptor getColumnDescriptor(List<FieldSchema> cols, long
tblId)
+ throws MetaException {
+
+ if (cols == null || cols.isEmpty()) {
+ return null;
+ }
+
+ // Please note! In case you modify any of those methods, run
TestHMSColumnDescriptorReuse with all the supported databases
+ List<Long> cdCandidates = findTheLatestColumnDescriptors(tblId);
+
+ cdCandidates = filterCandidatesByColumnCount(cdCandidates, cols.size());
+
+ if (cdCandidates == null || cdCandidates.isEmpty()) {
+ return null;
+ }
+
+ Long matchedColumnDescriptorId =
matchColumnDescriptorWithActualColumns(cdCandidates, cols);
+
+ if (matchedColumnDescriptorId != null) {
+ return pm.getObjectById(MColumnDescriptor.class,
matchedColumnDescriptorId);
+ }
+
+ return null;
+ }
+
+ private List<Long> findTheLatestColumnDescriptors(long tblId) throws
MetaException {
+ final int limit = 20;
Review Comment:
PR description says "Take the latest 10 partition", but the implementation
uses a limit of 20. Please align the code with the described algorithm (or
update the PR description to match the implementation).
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java:
##########
@@ -3331,26 +3332,85 @@ private 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());
- }
+ MStorageDescriptor msd = convertToMStorageDescriptor(part.getSd(), mt);
return new MPartition(Warehouse.makePartName(convertToFieldSchemas(mt
.getPartitionKeys()), part.getValues()), mt, part.getValues(), part
.getCreateTime(), part.getLastAccessTime(),
msd, part.getParameters());
}
+ /**
+ * 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 MStorageDescriptor convertToMStorageDescriptor(StorageDescriptor sd,
MTable mt)
+ throws MetaException {
+ if (sd == null) {
+ return null;
+ }
+
+ MColumnDescriptor mcd = (mt != null) ? getColumnDescriptor(sd.getCols(),
mt) : null;
+ if (mcd == null) {
+ mcd = createNewMColumnDescriptor(convertToMFieldSchemas(sd.getCols()));
+ }
+ return convertToMStorageDescriptor(sd, mcd);
+ }
+
+ 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();
+ }
+
+ if (!MetastoreConf.getBoolVar(conf, PARTITION_REUSE_COLUMN_DESCRIPTORS)) {
+ return null;
+ }
+
+ String catName = mt.getDatabase().getCatalogName();
+ String dbName = mt.getDatabase().getName();
+ String tblName = mt.getTableName();
+ long tblId = mt.getId();
+ try {
+ return new GetHelper<TableName, MColumnDescriptor>(this, new
TableName(catName, dbName, tblName)) {
Review Comment:
These variables are only used to construct a TableName passed to GetHelper,
but run(false) never initializes the table so the argument is unused. This adds
unnecessary work and also risks an NPE if mt.getDatabase() is unexpectedly
null. You can pass null as the GetHelper argument here and drop the extra
lookups.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/metastore/impl/TableStoreImpl.java:
##########
@@ -3331,26 +3332,85 @@ private 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());
- }
+ MStorageDescriptor msd = convertToMStorageDescriptor(part.getSd(), mt);
return new MPartition(Warehouse.makePartName(convertToFieldSchemas(mt
.getPartitionKeys()), part.getValues()), mt, part.getValues(), part
.getCreateTime(), part.getLastAccessTime(),
msd, part.getParameters());
}
+ /**
+ * 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 MStorageDescriptor convertToMStorageDescriptor(StorageDescriptor sd,
MTable mt)
+ throws MetaException {
+ if (sd == null) {
+ return null;
+ }
+
+ MColumnDescriptor mcd = (mt != null) ? getColumnDescriptor(sd.getCols(),
mt) : null;
+ if (mcd == null) {
+ mcd = createNewMColumnDescriptor(convertToMFieldSchemas(sd.getCols()));
+ }
+ return convertToMStorageDescriptor(sd, mcd);
+ }
+
+ 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
Review Comment:
Typos in this comment (e.g., "tables", "fast-check", "round-tripe") make it
harder to read and search. Please fix the wording.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
