You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by ta...@apache.org on 2018/12/26 15:33:00 UTC
impala git commit: IMPALA-7047. Refreshing partitions should not make
an RPC per file
Repository: impala
Updated Branches:
refs/heads/master 32fc07b50 -> c59f76186
IMPALA-7047. Refreshing partitions should not make an RPC per file
The code to handle REFRESH of a single partition was incorrectly
ignoring the previously-known file descriptors. This meant that, instead
of only calling 'getFileBlockLocations' on the files that had changed
since the prior load, it was instead calling it on every file.
In addition to refresh of single partitions this also affected refresh
of unpartitioned tables (which is implemented as a refresh of its single
"default" partition).
This patch fixes the behavior by copying over the existing file
descriptor list into the re-created partition before refreshing it. A
new unit test uses FS statistics to verify the change. The new
assertions act as a regression test and fail if I comment out the fix.
Additionally, this fixes the case where the old partition had no files
to use the optimized 'listLocatedStatus' call. This is triggered when
'REFRESH' picks up a new partition from the HMS added by an external
system.
I also tested this by pointing my dev box at a remote filesystem that
was approximately 60ms away. The initial load of an unpartitioned table
with approximately 45000 files takes around 23 seconds in this setup.
Without the patch in place, REFRESH was taking upwards of 35 minutes (I
got tired and gave up at this point). Multiplying the 60ms round trip by
45000 files estimates 45 minutes. With the fix in place, REFRESH of the
same table took around 4.5 seconds.
Clearly, in typical setups where catalogd and HDFS are on a shared local
network, the gains won't be so dramatic. But, even with a 1ms round trip
(plausible when including fixed RPC overhead and potentially congested
datacenter networks) this would save 45 seconds on this example table
with 45000 files.
Change-Id: I2051b96599206164aaa06ecbdf64374c46eda956
Reviewed-on: http://gerrit.cloudera.org:8080/11227
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/impala/repo
Commit: http://git-wip-us.apache.org/repos/asf/impala/commit/c59f7618
Tree: http://git-wip-us.apache.org/repos/asf/impala/tree/c59f7618
Diff: http://git-wip-us.apache.org/repos/asf/impala/diff/c59f7618
Branch: refs/heads/master
Commit: c59f76186aa98c1bd418d2d2951561f0521edec8
Parents: 32fc07b
Author: Todd Lipcon <to...@apache.org>
Authored: Tue Aug 14 22:34:23 2018 -0700
Committer: Impala Public Jenkins <im...@cloudera.com>
Committed: Fri Dec 21 23:23:56 2018 +0000
----------------------------------------------------------------------
.../apache/impala/catalog/HdfsPartition.java | 1 +
.../org/apache/impala/catalog/HdfsTable.java | 43 +++++++++---
.../org/apache/impala/catalog/CatalogTest.java | 70 +++++++++++++++++---
3 files changed, 97 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/impala/blob/c59f7618/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
index cd52b5b..e497b86 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
@@ -730,6 +730,7 @@ public class HdfsPartition implements FeFsPartition, PrunablePartition {
return encodedFileDescriptors_.size();
}
+ @Override
public boolean hasFileDescriptors() { return !encodedFileDescriptors_.isEmpty(); }
// Struct-style class for caching all the information we need to reconstruct an
http://git-wip-us.apache.org/repos/asf/impala/blob/c59f7618/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
----------------------------------------------------------------------
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
index 768e0bf..f2fd897 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -561,11 +561,17 @@ public class HdfsTable extends Table implements FeFsTable {
throws CatalogException {
try {
Path partDir = partition.getLocationPath();
- FileMetadataLoadStats stats = refreshFileMetadata(partDir,
- Lists.newArrayList(partition));
+ // If there are no existing files in the partition, use the non-incremental path.
+ boolean useExistingFds = partition.hasFileDescriptors();
+ FileMetadataLoadStats stats;
+ if (useExistingFds) {
+ stats = refreshFileMetadata(partDir, Collections.singletonList(partition));
+ } else {
+ stats = resetAndLoadFileMetadata(partDir, Collections.singletonList(partition));
+ }
if (LOG.isDebugEnabled()) {
- LOG.debug(String.format("Refreshed file metadata for %s %s",
- getFullName(), stats.debugString()));
+ LOG.debug("{} file metadata for {} {}", useExistingFds ? "Refreshed" : "Loaded",
+ getFullName(), stats.debugString());
}
} catch (IOException e) {
throw new CatalogException("Encountered invalid partition path", e);
@@ -852,9 +858,10 @@ public class HdfsTable extends Table implements FeFsTable {
if (numPathsToLoad == 0) return;
int threadPoolSize = getLoadingThreadPoolSize(numPathsToLoad);
- LOG.info(String.format("Loading file and block metadata for %s paths for table %s " +
- "using a thread pool of size %s", numPathsToLoad, getFullName(),
- threadPoolSize));
+ LOG.info("{} file and block metadata for {} paths for table {} " +
+ "using a thread pool of size {}",
+ reuseFileMd ? "Refreshing" : "Loading", numPathsToLoad, getFullName(),
+ threadPoolSize);
ExecutorService partitionLoadingPool = Executors.newFixedThreadPool(threadPoolSize);
try {
List<Future<FileMetadataLoadStats>> pendingMdLoadTasks = Lists.newArrayList();
@@ -1262,18 +1269,30 @@ public class HdfsTable extends Table implements FeFsTable {
}
/**
- * Updates the file metadata of an unpartitioned HdfsTable.
+ * Incrementally updates the file metadata of an unpartitioned HdfsTable.
+ *
+ * This is optimized for the case where few files have changed. See
+ * {@link #refreshFileMetadata()} above for details.
*/
private void updateUnpartitionedTableFileMd() throws CatalogException {
Preconditions.checkState(getNumClusteringCols() == 0);
if (LOG.isTraceEnabled()) {
LOG.trace("update unpartitioned table: " + getFullName());
}
+ HdfsPartition oldPartition = Iterables.getOnlyElement(partitionMap_.values());
+
+ // Instead of updating the existing partition in place, we create a new one
+ // so that we reflect any changes in the msTbl object and also assign a new
+ // ID. This is one step towards eventually implementing IMPALA-7533.
resetPartitions();
org.apache.hadoop.hive.metastore.api.Table msTbl = getMetaStoreTable();
Preconditions.checkNotNull(msTbl);
setPrototypePartition(msTbl.getSd());
HdfsPartition part = createPartition(msTbl.getSd(), null, new FsPermissionCache());
+ // Copy over the FDs from the old partition to the new one, so that
+ // 'refreshPartitionFileMetadata' below can compare modification times and
+ // reload the locations only for those that changed.
+ part.setFileDescriptors(oldPartition.getFileDescriptors());
addPartition(part);
if (isMarkedCached_) part.markCached();
refreshPartitionFileMetadata(part);
@@ -2138,11 +2157,17 @@ public class HdfsTable extends Table implements FeFsTable {
*/
public void reloadPartition(HdfsPartition oldPartition, Partition hmsPartition)
throws CatalogException {
+ // Instead of updating the existing partition in place, we create a new one
+ // so that we reflect any changes in the hmsPartition object and also assign a new
+ // ID. This is one step towards eventually implementing IMPALA-7533.
HdfsPartition refreshedPartition = createPartition(
hmsPartition.getSd(), hmsPartition, new FsPermissionCache());
- refreshPartitionFileMetadata(refreshedPartition);
Preconditions.checkArgument(oldPartition == null
|| HdfsPartition.KV_COMPARATOR.compare(oldPartition, refreshedPartition) == 0);
+ if (oldPartition != null) {
+ refreshedPartition.setFileDescriptors(oldPartition.getFileDescriptors());
+ }
+ refreshPartitionFileMetadata(refreshedPartition);
dropPartition(oldPartition, false);
addPartition(refreshedPartition);
}
http://git-wip-us.apache.org/repos/asf/impala/blob/c59f7618/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
----------------------------------------------------------------------
diff --git a/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java b/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
index 6b47bda..2226dc0 100644
--- a/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/CatalogTest.java
@@ -43,6 +43,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.GlobalStorageStatistics;
import org.apache.hadoop.fs.StorageStatistics;
import org.apache.hadoop.hdfs.DFSOpsCountStatistics;
+import org.apache.hadoop.hdfs.DFSOpsCountStatistics.OpType;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.metastore.api.ColumnStatisticsData;
import org.apache.impala.analysis.FunctionName;
@@ -56,6 +57,7 @@ import org.apache.impala.testutil.CatalogServiceTestCatalog;
import org.apache.impala.thrift.TBackendGflags;
import org.apache.impala.thrift.TFunctionBinaryType;
import org.apache.impala.thrift.TGetPartitionStatsRequest;
+import org.apache.impala.thrift.TPartitionKeyValue;
import org.apache.impala.thrift.TPartitionStats;
import org.apache.impala.thrift.TPrincipalType;
import org.apache.impala.thrift.TPrivilege;
@@ -66,6 +68,7 @@ import org.junit.Before;
import org.junit.Test;
import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
@@ -312,9 +315,20 @@ public class CatalogTest {
catalog_.getOrLoadTable("functional", "AllTypes"));
}
+ // Count of listFiles (list status + blocks) calls
+ private static final String LIST_LOCATED_STATUS =
+ OpType.LIST_LOCATED_STATUS.getSymbol();
+ // Count of listStatus calls
+ private static final String LIST_STATUS = OpType.LIST_STATUS.getSymbol();
+ // Count of getStatus calls
+ private static final String GET_FILE_STATUS = OpType.GET_FILE_STATUS.getSymbol();
+ // Count of getFileBlockLocations() calls
+ private static final String GET_FILE_BLOCK_LOCS =
+ OpType.GET_FILE_BLOCK_LOCATIONS.getSymbol();
+
/**
- * Regression test for IMPALA-7320: we should use batch APIs to fetch
- * file permissions for partitions.
+ * Regression test for IMPALA-7320 and IMPALA-7047: we should use batch APIs to fetch
+ * file permissions for partitions when loading or reloading.
*/
@Test
public void testNumberOfGetFileStatusCalls() throws CatalogException, IOException {
@@ -337,25 +351,65 @@ public class CatalogTest {
// Due to HDFS-13747, the listStatus calls are incorrectly accounted as
// op_list_located_status. So, we'll just add up the two to make our
// assertion resilient to this bug.
- long seenCalls = opsCounts.getLong("op_list_located_status") +
- opsCounts.getLong("op_list_status");
+ long seenCalls = opsCounts.getLong(LIST_LOCATED_STATUS) +
+ opsCounts.getLong(LIST_STATUS);
assertEquals(expectedCalls, seenCalls);
// We expect only one getFileStatus call, for the top-level directory.
- assertEquals(1L, (long)opsCounts.getLong("op_get_file_status"));
+ assertEquals(1L, (long)opsCounts.getLong(GET_FILE_STATUS));
+
+ // None of the underlying files changed so we should not do any ops for the files.
+ assertEquals(0L, (long)opsCounts.getLong(GET_FILE_BLOCK_LOCS));
// Now test REFRESH on the table...
stats.reset();
catalog_.reloadTable(table);
// Again, we expect only one getFileStatus call, for the top-level directory.
- assertEquals(1L, (long)opsCounts.getLong("op_get_file_status"));
+ assertEquals(1L, (long)opsCounts.getLong(GET_FILE_STATUS));
// REFRESH calls listStatus on each of the partitions, but doesn't re-check
// the permissions of the partition directories themselves.
assertEquals(table.getPartitionIds().size(),
- (long)opsCounts.getLong("op_list_status"));
- }
+ (long)opsCounts.getLong(LIST_STATUS));
+ // None of the underlying files changed so we should not do any ops for the files.
+ assertEquals(0L, (long)opsCounts.getLong(GET_FILE_BLOCK_LOCS));
+ // Reloading a specific partition should not make an RPC per file
+ // (regression test for IMPALA-7047).
+ stats.reset();
+ List<TPartitionKeyValue> partitionSpec = ImmutableList.of(
+ new TPartitionKeyValue("year", "2010"),
+ new TPartitionKeyValue("month", "10"));
+ catalog_.reloadPartition(table, partitionSpec);
+ assertEquals(0L, (long)opsCounts.getLong(GET_FILE_BLOCK_LOCS));
+
+ // Loading or reloading an unpartitioned table with some files in it should not make
+ // an RPC per file.
+ stats.reset();
+ HdfsTable unpartTable = (HdfsTable)catalog_.getOrLoadTable(
+ "functional", "alltypesaggmultifilesnopart");
+ assertEquals(0L, (long)opsCounts.getLong(GET_FILE_BLOCK_LOCS));
+ stats.reset();
+ catalog_.reloadTable(unpartTable);
+ assertEquals(0L, (long)opsCounts.getLong(GET_FILE_BLOCK_LOCS));
+
+ // Simulate an empty partition, which will trigger the full
+ // reload path. Since we can't modify HDFS itself via these tests, we
+ // do the next best thing: modify the metadata to revise history as
+ // though the partition used above were actually empty.
+ HdfsPartition hdfsPartition = table
+ .getPartitionFromThriftPartitionSpec(partitionSpec);
+ hdfsPartition.setFileDescriptors(new ArrayList<>());
+ stats.reset();
+ catalog_.reloadPartition(table, partitionSpec);
+
+ // Should not scan the directory file-by-file, should use a single
+ // listLocatedStatus() to get the whole directory (partition)
+ assertEquals(0L, (long)opsCounts.getLong(GET_FILE_BLOCK_LOCS));
+ seenCalls = opsCounts.getLong(LIST_LOCATED_STATUS) +
+ opsCounts.getLong(LIST_STATUS);
+ assertEquals(1, seenCalls);
+ }
@Test
public void TestPartitions() throws CatalogException {