okumin commented on code in PR #5886:
URL: https://github.com/apache/hive/pull/5886#discussion_r2173178050
##########
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorOnTezTest.java:
##########
@@ -541,4 +543,23 @@ protected void dropTable(String tblName) throws Exception {
executeStatementOnDriver("drop table " + tblName, driver);
}
}
+
+ protected void startInitiator() throws Exception {
+
runOneLoopOfCompactorThread(CompactorTestUtilities.CompactorThreadType.INITIATOR);
+ }
+
+ private void
runOneLoopOfCompactorThread(CompactorTestUtilities.CompactorThreadType type)
throws Exception {
Review Comment:
Do we need this method?
##########
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Initiator.java:
##########
@@ -428,4 +286,24 @@ public void run() {
public void enforceMutex(boolean enableMutex) {
this.shouldUseMutex = enableMutex;
}
+
+ public Partition resolvePartition(CompactionInfo ci) throws Exception {
Review Comment:
```suggestion
private Partition resolvePartition(CompactionInfo ci) throws Exception {
```
##########
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/AcidTableOptimizer.java:
##########
@@ -0,0 +1,110 @@
+/*
+ * 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.ql.txn.compactor;
+
+import org.apache.hadoop.hive.common.ValidTxnList;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.Warehouse;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.ShowCompactResponse;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.txn.TxnCommonUtils;
+import org.apache.hadoop.hive.metastore.txn.TxnStore;
+import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
+import org.apache.hadoop.hive.ql.io.AcidUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+public class AcidTableOptimizer extends TableOptimizer {
+ static final private String CLASS_NAME = AcidTableOptimizer.class.getName();
+ static final private Logger LOG = LoggerFactory.getLogger(CLASS_NAME);
+
+ public AcidTableOptimizer(HiveConf conf, TxnStore txnHandler, MetadataCache
metadataCache) {
+ super(conf, txnHandler, metadataCache);
+ }
+
+ @Override
+ public Set<CompactionInfo> findPotentialCompactions(long lastChecked,
ShowCompactResponse currentCompactions,
+ Set<String> skipDBs, Set<String> skipTables) throws MetaException {
+
+ int abortedThreshold = HiveConf.getIntVar(conf,
+ HiveConf.ConfVars.HIVE_COMPACTOR_ABORTEDTXN_THRESHOLD);
+ long abortedTimeThreshold = HiveConf
+ .getTimeVar(conf,
HiveConf.ConfVars.HIVE_COMPACTOR_ABORTEDTXN_TIME_THRESHOLD,
+ TimeUnit.MILLISECONDS);
+
+ Set<CompactionInfo> potentials =
txnHandler.findPotentialCompactions(abortedThreshold, abortedTimeThreshold,
lastChecked)
+ .parallelStream()
+ .filter(ci -> isEligibleForCompaction(ci, currentCompactions,
skipDBs, skipTables))
+ .collect(Collectors.toSet());
+
+ if (!potentials.isEmpty()) {
+ ValidTxnList validTxnList = TxnCommonUtils.createValidReadTxnList(
+ txnHandler.getOpenTxns(), 0);
+ conf.set(ValidTxnList.VALID_TXNS_KEY, validTxnList.writeToString());
+ }
+
+ return potentials;
+ }
+
Review Comment:
```suggestion
@Override
```
From SonarQube
##########
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/TableOptimizer.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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.ql.txn.compactor;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.Warehouse;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.ShowCompactResponse;
+import org.apache.hadoop.hive.metastore.api.ShowCompactResponseElement;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
+import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
+import org.apache.hadoop.hive.metastore.txn.TxnStore;
+import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
+
+import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils;
+import org.apache.hadoop.hive.ql.exec.repl.util.ReplUtils;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Comparator;
+import java.util.List;
+import java.util.LongSummaryStatistics;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import static org.apache.hadoop.hive.metastore.HMSHandler.getMSForConf;
+import static
org.apache.hadoop.hive.metastore.utils.MetaStoreUtils.getDefaultCatalog;
+
+public abstract class TableOptimizer {
+ static final private String CLASS_NAME = AcidTableOptimizer.class.getName();
+ static final private Logger LOG = LoggerFactory.getLogger(CLASS_NAME);
+
+ public abstract Set<CompactionInfo> findPotentialCompactions(long
lastChecked, ShowCompactResponse currentCompactions,
+ Set<String> skipDBs, Set<String> skipTables) throws MetaException;
+
+ protected final HiveConf conf;
+ protected final TxnStore txnHandler;
+ protected final MetadataCache metadataCache;
+
+ public TableOptimizer(HiveConf conf, TxnStore txnHandler, MetadataCache
metadataCache) {
+ this.conf = conf;
+ this.txnHandler = txnHandler;
+ this.metadataCache = metadataCache;
+ }
+
+ protected boolean isEligibleForCompaction(CompactionInfo ci,
ShowCompactResponse currentCompactions,
+ Set<String> skipDBs, Set<String> skipTables) {
+ try {
+ if (skipDBs.contains(ci.dbname)) {
+ LOG.info("Skipping {}::{}, skipDBs::size:{}", ci.dbname, ci.tableName,
skipDBs.size());
+ return false;
+ } else {
+ if (replIsCompactionDisabledForDatabase(ci.dbname)) {
+ skipDBs.add(ci.dbname);
+ LOG.info("Skipping {} as compaction is disabled due to repl;
skipDBs::size:{}",
+ ci.dbname, skipDBs.size());
+ return false;
+ }
+ }
+
+ if (skipTables.contains(ci.getFullTableName())) {
+ return false;
+ }
+
+ LOG.info("Checking to see if we should compact {}",
ci.getFullPartitionName());
+
+ // Check if we have already initiated or are working on a compaction for
this table/partition.
+ // Also make sure we haven't exceeded configured number of consecutive
failures.
+ // If any of the above applies, skip it.
+ // Note: if we are just waiting on cleaning we can still check, as it
may be time to compact again even though we haven't cleaned.
+ if (foundCurrentOrFailedCompactions(currentCompactions, ci)) {
+ return false;
+ }
+
+ Table t = metadataCache.computeIfAbsent(ci.getFullTableName(), () ->
+ CompactorUtil.resolveTable(conf, ci.dbname, ci.tableName));
+ if (t == null) {
+ LOG.info("Can't find table {}, assuming it's a temp table or has been
dropped and moving on.",
+ ci.getFullTableName());
+ return false;
+ }
+
+ if (replIsCompactionDisabledForTable(t)) {
+ skipTables.add(ci.getFullTableName());
+ return false;
+ }
+
+ Map<String, String> dbParams = metadataCache.computeIfAbsent(ci.dbname,
() -> resolveDatabase(ci)).getParameters();
+ if (MetaStoreUtils.isNoAutoCompactSet(dbParams, t.getParameters())) {
+ if (Boolean.parseBoolean(MetaStoreUtils.getNoAutoCompact(dbParams))) {
+ skipDBs.add(ci.dbname);
+ LOG.info("DB {} marked {}=true so we will not compact it.",
hive_metastoreConstants.NO_AUTO_COMPACT, ci.dbname);
+ } else {
+ skipTables.add(ci.getFullTableName());
+ LOG.info("Table {} marked {}=true so we will not compact it.",
hive_metastoreConstants.NO_AUTO_COMPACT,
+ Warehouse.getQualifiedName(t));
+ }
+ return false;
+ }
+ } catch (Throwable e) {
+ LOG.error("Caught exception while checking compaction eligibility.", e);
+ try {
+ ci.errorMessage = e.getMessage();
+ txnHandler.markFailed(ci);
+ } catch (MetaException ex) {
+ LOG.error("Caught exception while marking compaction as failed.", e);
+ }
+ return false;
+ }
+ return true;
+ }
+
+ private boolean replIsCompactionDisabledForTable(Table tbl) {
+ // Compaction is disabled until after first successful incremental load.
Check HIVE-21197 for more detail.
+ boolean isCompactDisabled =
ReplUtils.isFirstIncPending(tbl.getParameters());
+ if (isCompactDisabled) {
+ LOG.info("Compaction is disabled for table {}", tbl.getTableName());
+ }
+ return isCompactDisabled;
+ }
+
+ private boolean replIsCompactionDisabledForDatabase(String dbName) throws
TException {
+ try {
+ Database database =
getMSForConf(conf).getDatabase(getDefaultCatalog(conf), dbName);
+ // Compaction is disabled until after first successful incremental load.
Check HIVE-21197 for more detail.
+ boolean isReplCompactDisabled =
ReplUtils.isFirstIncPending(database.getParameters());
+ if (isReplCompactDisabled) {
+ LOG.info("Compaction is disabled for database {}", dbName);
+ }
+ return isReplCompactDisabled;
+ } catch (NoSuchObjectException e) {
+ LOG.info("Unable to find database {}", dbName);
+ return true;
+ }
+ }
+
+ protected boolean foundCurrentOrFailedCompactions(ShowCompactResponse
compactions, CompactionInfo ci) throws MetaException {
+ if (compactions.getCompacts() == null) {
+ return false;
+ }
+
+ //In case of an aborted Dynamic partition insert, the created entry in the
compaction queue does not contain
+ //a partition name even for partitioned tables. As a result it can happen
that the ShowCompactResponse contains
+ //an element without partition name for partitioned tables. Therefore, it
is necessary to null check the partition
+ //name of the ShowCompactResponseElement even if the
CompactionInfo.partName is not null. These special compaction
+ //requests are skipped by the worker, and only cleaner will pick them up,
so we should allow to schedule a 'normal'
+ //compaction for partitions of those tables which has special (DP abort)
entry with undefined partition name.
+ List<ShowCompactResponseElement> filteredElements =
compactions.getCompacts().stream()
+ .filter(e -> e.getDbname().equals(ci.dbname)
+ && e.getTablename().equals(ci.tableName)
+ && (e.getPartitionname() == null && ci.partName == null ||
+ (Objects.equals(e.getPartitionname(),ci.partName))))
+ .toList();
+
+ // Figure out if there are any currently running compactions on the same
table or partition.
+ if (filteredElements.stream().anyMatch(
+ e -> TxnStore.WORKING_RESPONSE.equals(e.getState()) ||
TxnStore.INITIATED_RESPONSE.equals(e.getState()))) {
+
+ LOG.info("Found currently initiated or working compaction for {} so we
will not initiate another compaction",
+ ci.getFullPartitionName());
+ return true;
+ }
+
+ // Check if there is already sufficient number of consecutive failures for
this table/partition
+ // so that no new automatic compactions needs to be scheduled.
+ int failedThreshold = MetastoreConf.getIntVar(conf,
MetastoreConf.ConfVars.COMPACTOR_INITIATOR_FAILED_THRESHOLD);
+
+ LongSummaryStatistics failedStats = filteredElements.stream()
+ .filter(e -> TxnStore.SUCCEEDED_RESPONSE.equals(e.getState()) ||
TxnStore.FAILED_RESPONSE.equals(e.getState()))
+
.sorted(Comparator.comparingLong(ShowCompactResponseElement::getId).reversed())
+ .limit(failedThreshold)
+
+ .filter(e -> TxnStore.FAILED_RESPONSE.equals(e.getState()))
+
.collect(Collectors.summarizingLong(ShowCompactResponseElement::getEnqueueTime));
+
+ // If the last attempt was too long ago, ignore the failed threshold and
try compaction again
+ long retryTime = MetastoreConf.getTimeVar(conf,
+ MetastoreConf.ConfVars.COMPACTOR_INITIATOR_FAILED_RETRY_TIME,
TimeUnit.MILLISECONDS);
+
+ boolean needsRetry = (retryTime > 0) && (failedStats.getMax() + retryTime
< System.currentTimeMillis());
+ if (failedStats.getCount() == failedThreshold && !needsRetry) {
+ LOG.warn("Will not initiate compaction for {} since last {} attempts to
compact it failed.",
+ ci.getFullPartitionName(),
MetastoreConf.ConfVars.COMPACTOR_INITIATOR_FAILED_THRESHOLD);
+
+ ci.errorMessage = "Compaction is not initiated since last " +
+ MetastoreConf.ConfVars.COMPACTOR_INITIATOR_FAILED_THRESHOLD + "
consecutive compaction attempts failed)";
+
+ txnHandler.markFailed(ci);
+ return true;
+ }
+ return false;
+ }
+
+ protected Database resolveDatabase(CompactionInfo ci) throws MetaException,
NoSuchObjectException {
+ return CompactorUtil.resolveDatabase(conf, ci.dbname);
+ }
+
+ protected boolean isCacheEnabled() {
+ return MetastoreConf.getBoolVar(conf,
+ MetastoreConf.ConfVars.COMPACTOR_INITIATOR_TABLECACHE_ON);
+ }
Review Comment:
Is this used?
##########
itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestIcebergCompactorOnTez.java:
##########
@@ -21,66 +21,139 @@
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.metastore.api.CompactionType;
import org.apache.hadoop.hive.metastore.api.ShowCompactResponse;
+import org.apache.hadoop.hive.metastore.txn.TxnUtils;
+import org.apache.hadoop.hive.metastore.txn.entities.CompactionState;
import org.junit.Assert;
+import org.junit.Before;
import org.junit.Test;
import java.util.ArrayList;
import java.util.List;
+import java.util.Objects;
import static
org.apache.hadoop.hive.ql.txn.compactor.TestCompactor.executeStatementOnDriver;
public class TestIcebergCompactorOnTez extends CompactorOnTezTest {
+
+ private static final String DB_NAME = "default";
+ private static final String TABLE_NAME = "ice_orc";
+ private static final String QUALIFIED_TABLE_NAME =
TxnUtils.getFullTableName(DB_NAME, TABLE_NAME);
+
+ @Override
+ @Before
+ public void setup() throws Exception {
+ super.setup();
+ executeStatementOnDriver("drop table if exists " + QUALIFIED_TABLE_NAME,
driver);
+ }
@Test
public void testIcebergCompactorWithAllPartitionFieldTypes() throws
Exception{
conf.setVar(HiveConf.ConfVars.COMPACTOR_JOB_QUEUE,
CUSTOM_COMPACTION_QUEUE);
msClient = new HiveMetaStoreClient(conf);
- String dbName = "default";
- String tableName = "ice_orc";
- String qualifiedTableName = dbName + "." + tableName;
-
- executeStatementOnDriver("drop table if exists " + qualifiedTableName,
driver);
executeStatementOnDriver(String.format("create table %s " +
"(id int, a string, b int, c bigint, d float, e double, f decimal(4,
2), g boolean, h date, i date, j date, k timestamp) " +
"partitioned by spec(a, truncate(3, a), bucket(4, a), b, c, d, e, f,
g, h, year(h), month(i), day(j), k, hour(k)) stored by iceberg stored as orc " +
- "tblproperties ('compactor.threshold.min.input.files'='1')",
qualifiedTableName), driver);
+ "tblproperties ('compactor.threshold.min.input.files'='1')",
QUALIFIED_TABLE_NAME), driver);
// 6 records, one records per file --> 3 partitions, 2 files per partition
- executeStatementOnDriver(String.format("INSERT INTO %s VALUES (1,
'aaa111', 1, 100, 1.0, 2.0, 4.00, true, DATE '2024-05-01', DATE '2024-05-01',
DATE '2024-05-01', TIMESTAMP '2024-05-02 10:00:00')", qualifiedTableName),
driver);
- executeStatementOnDriver(String.format("INSERT INTO %s VALUES (2,
'aaa111', 1, 100, 1.0, 2.0, 4.00, true, DATE '2024-05-01', DATE '2024-05-01',
DATE '2024-05-01', TIMESTAMP '2024-05-02 10:00:00')", qualifiedTableName),
driver);
- executeStatementOnDriver(String.format("INSERT INTO %s VALUES (3,
'bbb222', 2, 200, 2.0, 3.0, 8.00, false, DATE '2024-05-03', DATE '2024-05-03',
DATE '2024-05-03', TIMESTAMP '2024-05-04 13:00:00')", qualifiedTableName),
driver);
- executeStatementOnDriver(String.format("INSERT INTO %s VALUES (4,
'bbb222', 2, 200, 2.0, 3.0, 8.00, false, DATE '2024-05-03', DATE '2024-05-03',
DATE '2024-05-03', TIMESTAMP '2024-05-04 13:00:00')", qualifiedTableName),
driver);
- executeStatementOnDriver(String.format("INSERT INTO %s VALUES (5, null,
null, null, null, null, null, null, null, null, null, null)",
qualifiedTableName), driver);
- executeStatementOnDriver(String.format("INSERT INTO %s VALUES (6, null,
null, null, null, null, null, null, null, null, null, null)",
qualifiedTableName), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (1,
'aaa111', 1, 100, 1.0, 2.0, 4.00, true, DATE '2024-05-01', DATE '2024-05-01',
DATE '2024-05-01', TIMESTAMP '2024-05-02 10:00:00')", QUALIFIED_TABLE_NAME),
driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (2,
'aaa111', 1, 100, 1.0, 2.0, 4.00, true, DATE '2024-05-01', DATE '2024-05-01',
DATE '2024-05-01', TIMESTAMP '2024-05-02 10:00:00')", QUALIFIED_TABLE_NAME),
driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (3,
'bbb222', 2, 200, 2.0, 3.0, 8.00, false, DATE '2024-05-03', DATE '2024-05-03',
DATE '2024-05-03', TIMESTAMP '2024-05-04 13:00:00')", QUALIFIED_TABLE_NAME),
driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (4,
'bbb222', 2, 200, 2.0, 3.0, 8.00, false, DATE '2024-05-03', DATE '2024-05-03',
DATE '2024-05-03', TIMESTAMP '2024-05-04 13:00:00')", QUALIFIED_TABLE_NAME),
driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (5, null,
null, null, null, null, null, null, null, null, null, null)",
QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (6, null,
null, null, null, null, null, null, null, null, null, null)",
QUALIFIED_TABLE_NAME), driver);
- Assert.assertEquals(6, getFilesCount(qualifiedTableName));
- List<String> recordsBefore = getAllRecords(qualifiedTableName);
+ Assert.assertEquals(6, getFilesCount());
+ List<String> recordsBefore = getAllRecords();
- CompactorTestUtil.runCompaction(conf, dbName, tableName,
CompactionType.MINOR, false,
+ CompactorTestUtil.runCompaction(conf, DB_NAME, TABLE_NAME,
CompactionType.MINOR, false,
"a=aaa111/a_trunc=aaa/a_bucket=0/b=1/c=100/d=1.0/e=2.0/f=4.00/g=true/h=2024-05-01/h_year=2024/i_month=2024-05/j_day=2024-05-01/k=2024-05-02T10%3A00%3A00/k_hour=2024-05-02-10",
"a=bbb222/a_trunc=bbb/a_bucket=3/b=2/c=200/d=2.0/e=3.0/f=8.00/g=false/h=2024-05-03/h_year=2024/i_month=2024-05/j_day=2024-05-03/k=2024-05-04T13%3A00%3A00/k_hour=2024-05-04-13",
"a=null/a_trunc=null/a_bucket=null/b=null/c=null/d=null/e=null/f=null/g=null/h=null/h_year=null/i_month=null/j_day=null/k=null/k_hour=null"
);
- Assert.assertEquals(3, getFilesCount(qualifiedTableName));
+ Assert.assertEquals(3, getFilesCount());
verifySuccessfulCompaction(3);
- List<String> recordsAfter = getAllRecords(qualifiedTableName);
+ List<String> recordsAfter = getAllRecords();
Assert.assertEquals(recordsBefore, recordsAfter);
}
-
- private int getFilesCount(String qualifiedTableName) throws Exception {
- driver.run(String.format("select count(*) from %s.files",
qualifiedTableName));
+
+ @Test
+ public void testIcebergAutoCompactionPartitionEvolution() throws Exception {
+ executeStatementOnDriver(String.format("create table %s " +
+ "(id int, a string) " +
+ "partitioned by spec(id) stored by iceberg stored as orc " +
+ "tblproperties ('compactor.threshold.min.input.files'='1')",
QUALIFIED_TABLE_NAME), driver);
+
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (1, 'a')",
QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (2, 'b')",
QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (3, 'c')",
QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (4, 'd')",
QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (5, 'e')",
QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (6, 'd')",
QUALIFIED_TABLE_NAME), driver);
+
+ executeStatementOnDriver(String.format("alter table %s set partition
spec(truncate(3, a))", QUALIFIED_TABLE_NAME), driver);
+
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (7,
'aaa111')", QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (8,
'aaa111')", QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (9,
'bbb222')", QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (10,
'bbb222')", QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (11, null)",
QUALIFIED_TABLE_NAME), driver);
+ executeStatementOnDriver(String.format("INSERT INTO %s VALUES (12, null)",
QUALIFIED_TABLE_NAME), driver);
+
+ startInitiator();
+ ShowCompactResponse rsp = msClient.showCompactions();
+ Assert.assertEquals(4, rsp.getCompactsSize());
+
+ // Compaction should be initiated for each partition from the latest spec
+ Assert.assertTrue(isCompactExist(rsp, "a_trunc_3=aaa",
CompactionType.MINOR, CompactionState.INITIATED));
+ Assert.assertTrue(isCompactExist(rsp, "a_trunc_3=bbb",
CompactionType.MINOR, CompactionState.INITIATED));
+ Assert.assertTrue(isCompactExist(rsp, "a_trunc_3=null",
CompactionType.MINOR, CompactionState.INITIATED));
Review Comment:
Although this is off-topic, I wonder what if there is a partition whose name
is "null" as a text.
--
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: gitbox-unsubscr...@hive.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscr...@hive.apache.org
For additional commands, e-mail: gitbox-h...@hive.apache.org
