| // 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.impala.util; |
| |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertThat; |
| |
| import java.util.Arrays; |
| import java.util.List; |
| import java.util.stream.Collectors; |
| |
| import org.apache.hadoop.fs.FileStatus; |
| import org.apache.hadoop.fs.Path; |
| import org.apache.hadoop.hive.common.ValidReadTxnList; |
| import org.apache.hadoop.hive.common.ValidWriteIdList; |
| import org.apache.impala.compat.MetastoreShim; |
| import org.hamcrest.Matchers; |
| import org.junit.Assume; |
| import org.junit.Test; |
| |
| public class AcidUtilsTest { |
| |
| /** Fake base path to root all FileStatuses under. */ |
| private static final Path BASE_PATH = new Path("file:///foo/bar/"); |
| |
| public AcidUtilsTest() { |
| Assume.assumeTrue("Tests require Hive 3 to parse and use WriteIdList", |
| MetastoreShim.getMajorVersion() == 3); |
| } |
| |
| private static List<FileStatus> createMockStats(String... testStrings) { |
| return Arrays.asList(testStrings).stream().map(s -> { |
| return new FileStatus( |
| /* length=*/0, /*isdir=*/s.endsWith("/"), /*block_replication=*/1, |
| /*blocksize=*/0, /*mtime=*/1, /*path=*/new Path(BASE_PATH, s)); |
| }).collect(Collectors.toList()); |
| } |
| |
| private void assertFiltering(String[] relPaths, String validWriteIdListStr, |
| String[] expectedRelPaths) { |
| assertFiltering(relPaths, "", validWriteIdListStr, expectedRelPaths); |
| } |
| |
| private void assertFiltering(String[] relPaths, String validTxnListStr, |
| String validWriteIdListStr, String[] expectedRelPaths) { |
| ValidWriteIdList writeIds = MetastoreShim.getValidWriteIdListFromString( |
| validWriteIdListStr); |
| List<FileStatus> stats = createMockStats(relPaths); |
| List<FileStatus> expectedStats = createMockStats(expectedRelPaths); |
| |
| assertThat(AcidUtils.filterFilesForAcidState(stats, BASE_PATH, |
| new ValidReadTxnList(validTxnListStr), writeIds, null), |
| Matchers.containsInAnyOrder(expectedStats.toArray())); |
| } |
| |
| @Test |
| public void testParsingBaseWriteIds() { |
| assertEquals(AcidUtils.SENTINEL_BASE_WRITE_ID, |
| AcidUtils.getBaseWriteId("base_01.txt")); |
| assertEquals(AcidUtils.SENTINEL_BASE_WRITE_ID, |
| AcidUtils.getBaseWriteId("base_01_02")); |
| assertEquals(AcidUtils.SENTINEL_BASE_WRITE_ID, |
| AcidUtils.getBaseWriteId("abc/base_6")); |
| assertEquals(2, AcidUtils.getBaseWriteId("base_00002")); |
| assertEquals(2, AcidUtils.getBaseWriteId("base_00002/foo")); |
| assertEquals(2, AcidUtils.getBaseWriteId("base_00002_v12345")); |
| assertEquals(2, AcidUtils.getBaseWriteId("base_00002_v12345/foo")); |
| } |
| |
| @Test |
| public void testBasePredicate() { |
| assertFiltering(new String[]{ |
| "base_01.txt", |
| "post_upgrade.txt", |
| "base_0000005/", |
| "base_0000005/abc.txt", |
| "base_0000005/0000/", |
| "base_0000005/0000/abc.txt", |
| "_tmp.base_0000005/000000_0.manifest", |
| "abc/", |
| "abc/base_0000006/", // Not at root, so shouldn't be handled. |
| "base_00000100/def.txt"}, |
| // <tableName>:<highWatermark>:<minOpenWriteId>:<openWriteIds>:<abortedWriteIds> |
| "default.test:10:1234:1,2,3", |
| new String[]{ |
| "base_0000005/abc.txt", |
| "base_0000005/0000/abc.txt"}); |
| } |
| |
| @Test |
| public void testOpenTransactions() { |
| assertFiltering(new String[]{ |
| "base_01.txt", |
| "post_upgrade.txt", |
| "base_0000005/", |
| "base_0000005/abc.txt", |
| "base_0000005/0000/", |
| "base_0000005/0000/abc.txt", |
| "_tmp.base_0000005/000000_0.manifest", |
| "delta_0000006_0000006_0000/", |
| "delta_0000006_0000006_0000/000000_0", |
| "delta_0000007_0000007_0000/", |
| "delta_0000007_0000007_0000/000000_0", |
| "delta_0000008_0000008_0000/", |
| "delta_0000008_0000008_0000/000000_0", |
| "delta_0000009_0000009_0000/", |
| "delta_0000009_0000009_0000/000000_0", |
| "delta_0000009_0000009_0000/0000/def.txt"}, |
| "default.test:10:6:6,7,8:", // 6,7,8 are open write ids |
| new String[]{ |
| "base_0000005/abc.txt", |
| "base_0000005/0000/abc.txt", |
| "delta_0000009_0000009_0000/000000_0", |
| "delta_0000009_0000009_0000/0000/def.txt"}); |
| } |
| |
| @Test |
| public void testInProgressCompaction() { |
| assertFiltering(new String[]{ |
| "base_01.txt", |
| "post_upgrade.txt", |
| "base_0000005_v900/", |
| "base_0000005_v900/abc.txt", |
| "base_0000005_v900/0000/", |
| "base_0000005_v900/0000/abc.txt", |
| "base_0000008_v1000/", |
| "base_0000008_v1000/abc.txt", |
| "delta_0000006_0000006_0000/", |
| "delta_0000006_0000006_0000/000000_0", |
| "delta_0000007_0000007_0000/", |
| "delta_0000007_0000007_0000/000000_0", |
| "delta_0000008_0000008_0000/", |
| "delta_0000008_0000008_0000/000000_0", |
| "delta_0000009_0000009_0000/", |
| "delta_0000009_0000009_0000/000000_0", |
| "delta_0000009_0000009_0000/0000/def.txt"}, |
| "1100:1000:1000:", // txn 1000 is open |
| "default.test:10:10::", // write ids are committed |
| new String[]{ |
| "base_0000005_v900/abc.txt", |
| "base_0000005_v900/0000/abc.txt", |
| "delta_0000006_0000006_0000/000000_0", |
| "delta_0000007_0000007_0000/000000_0", |
| "delta_0000008_0000008_0000/000000_0", |
| "delta_0000009_0000009_0000/000000_0", |
| "delta_0000009_0000009_0000/0000/def.txt"}); |
| } |
| |
| @Test |
| public void testAbortedCompaction() { |
| assertFiltering(new String[]{ |
| "base_01.txt", |
| "post_upgrade.txt", |
| "base_0000005_v900/", |
| "base_0000005_v900/abc.txt", |
| "base_0000005_v900/0000/", |
| "base_0000005_v900/0000/abc.txt", |
| "base_0000007_v950/", |
| "base_0000007_v950/abc.txt", |
| "base_0000008_v1000/", |
| "base_0000008_v1000/abc.txt", |
| "delta_0000006_0000006_0000/", |
| "delta_0000006_0000006_0000/000000_0", |
| "delta_0000007_0000007_0000/", |
| "delta_0000007_0000007_0000/000000_0", |
| "delta_0000008_0000008_0000/", |
| "delta_0000008_0000008_0000/000000_0", |
| "delta_0000009_0000009_0000/", |
| "delta_0000009_0000009_0000/000000_0", |
| "delta_0000009_0000009_0000/0000/def.txt"}, |
| "1100:1100::1000", // txn 1000 is aborted |
| "default.test:10:10::", // write ids are committed |
| new String[]{ |
| "base_0000007_v950/abc.txt", |
| "delta_0000008_0000008_0000/000000_0", |
| "delta_0000009_0000009_0000/000000_0", |
| "delta_0000009_0000009_0000/0000/def.txt"}); |
| } |
| |
| @Test |
| public void testAbortedTransactions() { |
| assertFiltering(new String[]{ |
| "base_01.txt", |
| "post_upgrade.txt", |
| "base_0000005/", |
| "base_0000005/abc.txt", |
| "base_0000005/0000/", |
| "base_0000005/0000/abc.txt", |
| "_tmp.base_0000005/000000_0.manifest", |
| "delta_0000006_0000006_0000/", |
| "delta_0000006_0000006_0000/000000_0", |
| "delta_0000007_0000007_0000/", |
| "delta_0000007_0000007_0000/000000_0", |
| "delta_0000008_0000008_0000/", |
| "delta_0000008_0000008_0000/000000_0", |
| "delta_0000009_0000009_0000/", |
| "delta_0000009_0000009_0000/000000_0", |
| "delta_0000009_0000009_0000/0000/def.txt"}, |
| "default.test:10:1337::7,8,9", // 7,8,9 are aborted write ids |
| new String[]{ |
| "base_0000005/abc.txt", |
| "base_0000005/0000/abc.txt", |
| "delta_0000006_0000006_0000/000000_0"}); |
| } |
| |
| @Test |
| public void testPostCompactionBase() { |
| assertFiltering(new String[]{ |
| "base_0000003_v0003217/", |
| "base_0000003_v0003217/000000_0", |
| "delta_0000001_0000001_0000/", |
| "delta_0000001_0000001_0000/000000_0", |
| "delta_0000002_0000002_0000/", |
| "delta_0000002_0000002_0000/000000_0", |
| "delta_0000003_0000003_0000/", |
| "delta_0000003_0000003_0000/000000_0"}, |
| "test_acid_compaction_676f9a30.tt:3:9223372036854775807::", |
| new String[]{ |
| "base_0000003_v0003217/000000_0"}); |
| } |
| |
| @Test |
| public void testDeltaPredicate() { |
| String[] paths = new String[]{ |
| "delta_000005_0000005/", |
| "delta_000005_0000005/abc.txt", |
| "delta_000005_0000005_0000/", |
| "delta_000005_0000005_0000/abc.txt", |
| "delta_000006_0000020/", |
| "delta_000006_0000020/def.txt", |
| "delta_000005.txt"}; |
| |
| // Only committed up to transaction 10, so skip the 6-20 delta. |
| assertFiltering(paths, |
| "default.test:10:1234:1,2,3", |
| new String[]{ |
| "delta_000005_0000005/abc.txt", |
| "delta_000005_0000005_0000/abc.txt", |
| "delta_000005.txt"}); |
| } |
| |
| @Test |
| public void testAcidStateFilter() { |
| assertFiltering(new String[]{ |
| "base_0000009/", |
| "base_0000009/abc.txt", |
| "delta_000005_000005_0000/", |
| "delta_000005_000005_0000/lmn.txt", |
| "base_0000010/", |
| "base_0000010/00000_0", |
| "base_0000010/00001_0", |
| "delta_0000012_0000012_0000/", |
| "delta_0000012_0000012_0000/0000_0", |
| "delta_0000012_0000012_0000/0000_1"}, |
| "", // Accept all, |
| new String[]{ |
| "base_0000010/00000_0", |
| "base_0000010/00001_0", |
| "delta_0000012_0000012_0000/0000_0", |
| "delta_0000012_0000012_0000/0000_1"}); |
| } |
| |
| @Test |
| public void testAcidStateNoBase() { |
| assertFiltering(new String[]{ |
| "base_01.txt", |
| "post_upgrade.txt", |
| "delta_000005_000005_0000/", |
| "delta_000005_000005_0000/lmn.txt", |
| "base_000010/", |
| "delta_0000012_0000012_0000/", |
| "delta_0000012_0000012_0000/0000_0", |
| "delta_0000012_0000012_0000/0000_1"}, |
| "", // writeIdList that accepts all transactions as valid |
| new String[]{ |
| // Post upgrade files are ignored if there is a valid base. |
| "delta_0000012_0000012_0000/0000_0", |
| "delta_0000012_0000012_0000/0000_1"}); |
| |
| // Same set of files, but no base directory. |
| assertFiltering(new String[]{ |
| "base_01.txt", |
| "post_upgrade.txt", |
| "delta_000005_000005_0000/", |
| "delta_000005_000005_0000/lmn.txt", |
| "delta_0000012_0000012_0000/", |
| "delta_0000012_0000012_0000/0000_0", |
| "delta_0000012_0000012_0000/0000_1"}, |
| "", // writeIdList that accepts all transactions as valid |
| new String[]{ |
| "base_01.txt", // Post upgrade files are considered valid if there is no base. |
| "post_upgrade.txt", |
| "delta_000005_000005_0000/lmn.txt", |
| "delta_0000012_0000012_0000/0000_0", |
| "delta_0000012_0000012_0000/0000_1"}); |
| } |
| } |
