You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by mi...@apache.org on 2025/02/27 22:45:10 UTC
(impala) branch master updated (9033e9aa0 -> 768527c89)
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 9033e9aa0 IMPALA-13781: Fix "Median Diff %" for report_benchmark_results.py
new a4f303ea5 IMPALA-13802: move postgresql init to the end of bin/bootstrap_system.sh
new 8c51f72e1 IMPALA-13789: Defer creating Path objects in loading file metadata
new 91343369f IMPALA-13785: Deflake test_catalog_tables_stats in exhaustive build
new 768527c89 IMPALA-13804: Use redacted statement in live table
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/exec/system-table-scanner.cc | 2 +-
bin/bootstrap_system.sh | 100 +++++++++++----------
.../apache/impala/catalog/CatalogHmsAPIHelper.java | 14 ++-
.../java/org/apache/impala/catalog/FeFsTable.java | 67 ++++++++++++++
.../apache/impala/catalog/FileMetadataLoader.java | 36 ++++----
.../org/apache/impala/catalog/HdfsPartition.java | 4 +-
.../java/org/apache/impala/catalog/HdfsTable.java | 65 --------------
.../impala/catalog/IcebergFileMetadataLoader.java | 30 ++++---
.../org/apache/impala/catalog/IcebergTable.java | 34 ++++++-
.../impala/catalog/ParallelFileMetadataLoader.java | 14 ++-
.../impala/catalog/local/DirectMetaProvider.java | 5 +-
.../org/apache/impala/common/FileSystemUtil.java | 15 ++++
.../impala/catalog/FileMetadataLoaderTest.java | 16 ++--
.../apache/impala/catalog/HdfsPartitionTest.java | 2 +-
testdata/workload_mgmt/redaction.json | 11 +++
tests/custom_cluster/test_query_live.py | 16 +++-
tests/custom_cluster/test_query_log.py | 23 ++++-
tests/util/workload_management.py | 7 ++
tests/webserver/test_web_pages.py | 23 +++++
19 files changed, 308 insertions(+), 176 deletions(-)
create mode 100644 testdata/workload_mgmt/redaction.json
(impala) 03/04: IMPALA-13785: Deflake test_catalog_tables_stats in exhaustive build
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 91343369f7f34c0990129d565b9172001ee5570b
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
AuthorDate: Mon Feb 24 18:22:43 2025 +0100
IMPALA-13785: Deflake test_catalog_tables_stats in exhaustive build
IMPALA-13737 changed how we load file metadata for Iceberg tables,
as we don't rely on the underlying HdfsTable anymore. As a result,
metrics were not correctly updated. This patch adds code to update
metrics correctly.
Testing:
* executed the tests in the title in exhaustive mode
* added Iceberg-specific tests to test_web_pages
Change-Id: I686e0e32b36ace4c5a8c9eeb715148c98521252a
Reviewed-on: http://gerrit.cloudera.org:8080/22531
Reviewed-by: Quanlong Huang <hu...@gmail.com>
Reviewed-by: Noemi Pap-Takacs <np...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Riza Suminto <ri...@cloudera.com>
---
.../java/org/apache/impala/catalog/FeFsTable.java | 67 ++++++++++++++++++++++
.../apache/impala/catalog/FileMetadataLoader.java | 2 +-
.../org/apache/impala/catalog/HdfsPartition.java | 4 +-
.../java/org/apache/impala/catalog/HdfsTable.java | 65 ---------------------
.../impala/catalog/IcebergFileMetadataLoader.java | 2 +-
.../org/apache/impala/catalog/IcebergTable.java | 34 ++++++++++-
tests/webserver/test_web_pages.py | 23 ++++++++
7 files changed, 127 insertions(+), 70 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java b/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
index 4edb8858f..b4d464b4f 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeFsTable.java
@@ -82,10 +82,77 @@ public interface FeFsTable extends FeTable {
// Internal table property that specifies the total size of the table.
public static final String TOTAL_SIZE = "totalSize";
+ // FS table specific metrics
+ public static final String NUM_FILES_METRIC = "num-files";
+ public static final String NUM_BLOCKS_METRIC = "num-blocks";
+ public static final String TOTAL_FILE_BYTES_METRIC = "total-file-size-bytes";
+ public static final String MEMORY_ESTIMATE_METRIC = "memory-estimate-bytes";
+
// Internal table property that specifies the number of erasure coded files in the
// table.
public static final String NUM_ERASURE_CODED_FILES = "numFilesErasureCoded";
+ // Average memory requirements (in bytes) for storing a file descriptor.
+ public static final long PER_FD_MEM_USAGE_BYTES = 500;
+
+ // Average memory requirements (in bytes) for storing a block.
+ public static final long PER_BLOCK_MEM_USAGE_BYTES = 150;
+
+ // Represents a set of storage-related statistics aggregated at the table or partition
+ // level.
+ public final static class FileMetadataStats {
+ // Number of files in a table/partition.
+ public long numFiles = 0;
+ // Number of blocks in a table/partition.
+ public long numBlocks = 0;
+ // Total size (in bytes) of all files in a table/partition.
+ public long totalFileBytes = 0;
+
+ public FileMetadataStats() {}
+
+ /**
+ * This constructor allows third party extensions to instantiate a FileMetadataStats
+ * with a List of FileDescriptor's.
+ */
+ public FileMetadataStats(List<FileDescriptor> fds) {
+ for (FileDescriptor fd : fds) {
+ accumulate(fd);
+ }
+ }
+
+ // Initializes the values of the storage stats.
+ public void init() {
+ numFiles = 0;
+ numBlocks = 0;
+ totalFileBytes = 0;
+ }
+
+ public void set(FileMetadataStats stats) {
+ numFiles = stats.numFiles;
+ numBlocks = stats.numBlocks;
+ totalFileBytes = stats.totalFileBytes;
+ }
+
+ public void merge(FileMetadataStats other) {
+ numFiles += other.numFiles;
+ numBlocks += other.numBlocks;
+ totalFileBytes += other.totalFileBytes;
+ }
+
+ public void remove(FileMetadataStats other) {
+ numFiles -= other.numFiles;
+ numBlocks -= other.numBlocks;
+ totalFileBytes -= other.totalFileBytes;
+ }
+
+ // Accumulate the statistics of the fd into this FileMetadataStats.
+ public void accumulate(FileDescriptor fd) {
+ numBlocks += fd.getNumFileBlocks();
+ totalFileBytes += fd.getFileLength();
+ ++numFiles;
+ }
+ }
+
/**
* @return true if the table and all its partitions reside at locations which
* support caching (e.g. HDFS).
diff --git a/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
index 214811241..fce49a9fe 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
@@ -30,7 +30,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.hive.common.ValidTxnList;
import org.apache.hadoop.hive.common.ValidWriteIdList;
-import org.apache.impala.catalog.HdfsTable.FileMetadataStats;
+import org.apache.impala.catalog.FeFsTable.FileMetadataStats;
import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.common.Reference;
import org.apache.impala.thrift.TNetworkAddress;
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 3d337beec..c3eef49fe 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsPartition.java
@@ -44,7 +44,7 @@ import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.LiteralExpr;
import org.apache.impala.analysis.PartitionKeyValue;
-import org.apache.impala.catalog.HdfsTable.FileMetadataStats;
+import org.apache.impala.catalog.FeFsTable.FileMetadataStats;
import org.apache.impala.catalog.events.InFlightEvents;
import org.apache.impala.catalog.events.MetastoreEvents.MetastoreEventPropertyKey;
import org.apache.impala.common.FileSystemUtil;
@@ -275,7 +275,7 @@ public class HdfsPartition extends CatalogObjectImpl
isMarkedCached, accessLevel, hmsParameters, cachedMsPartitionDescriptor,
partitionStats, hasIncrementalStats, numRows, writeId,
inFlightEvents, /*createEventId=*/-1L, /*lastCompactionId*/-1L,
- /*lastRefreshEventId*/-1L, new HdfsTable.FileMetadataStats());
+ /*lastRefreshEventId*/-1L, new FileMetadataStats());
}
/**
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 b68d21eb4..020df5e1d 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -175,19 +175,9 @@ public class HdfsTable extends Table implements FeFsTable {
// Average memory requirements (in bytes) for storing the metadata of a partition.
private static final long PER_PARTITION_MEM_USAGE_BYTES = 2048;
- // Average memory requirements (in bytes) for storing a file descriptor.
- private static final long PER_FD_MEM_USAGE_BYTES = 500;
-
- // Average memory requirements (in bytes) for storing a block.
- private static final long PER_BLOCK_MEM_USAGE_BYTES = 150;
-
// Hdfs table specific metrics
public static final String CATALOG_UPDATE_DURATION_METRIC = "catalog-update-duration";
public static final String NUM_PARTITIONS_METRIC = "num-partitions";
- public static final String NUM_FILES_METRIC = "num-files";
- public static final String NUM_BLOCKS_METRIC = "num-blocks";
- public static final String TOTAL_FILE_BYTES_METRIC = "total-file-size-bytes";
- public static final String MEMORY_ESTIMATE_METRIC = "memory-estimate-bytes";
public static final String HAS_INCREMENTAL_STATS_METRIC = "has-incremental-stats";
// metrics used to find out the cache hit rate when file-metadata is requested
// for a given ValidWriteIdList
@@ -343,61 +333,6 @@ public class HdfsTable extends Table implements FeFsTable {
skipIcebergFileMetadataLoading_ = skipIcebergFileMetadataLoading;
}
- // Represents a set of storage-related statistics aggregated at the table or partition
- // level.
- public final static class FileMetadataStats {
- // Number of files in a table/partition.
- public long numFiles = 0;
- // Number of blocks in a table/partition.
- public long numBlocks = 0;
- // Total size (in bytes) of all files in a table/partition.
- public long totalFileBytes = 0;
-
- public FileMetadataStats() {}
-
- /**
- * This constructor allows third party extensions to instantiate a FileMetadataStats
- * with a List of FileDescriptor's.
- */
- public FileMetadataStats(List<FileDescriptor> fds) {
- for (FileDescriptor fd : fds) {
- accumulate(fd);
- }
- }
-
- // Initializes the values of the storage stats.
- public void init() {
- numFiles = 0;
- numBlocks = 0;
- totalFileBytes = 0;
- }
-
- public void set(FileMetadataStats stats) {
- numFiles = stats.numFiles;
- numBlocks = stats.numBlocks;
- totalFileBytes = stats.totalFileBytes;
- }
-
- public void merge(FileMetadataStats other) {
- numFiles += other.numFiles;
- numBlocks += other.numBlocks;
- totalFileBytes += other.totalFileBytes;
- }
-
- public void remove(FileMetadataStats other) {
- numFiles -= other.numFiles;
- numBlocks -= other.numBlocks;
- totalFileBytes -= other.totalFileBytes;
- }
-
- // Accumulate the statistics of the fd into this FileMetadataStats.
- public void accumulate(FileDescriptor fd) {
- numBlocks += fd.getNumFileBlocks();
- totalFileBytes += fd.getFileLength();
- ++numFiles;
- }
- }
-
// Table level storage-related statistics. Depending on whether the table is stored in
// the catalog server or the impalad catalog cache, these statistics serve different
// purposes and, hence, are managed differently.
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
index 1b258467e..c77766723 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
@@ -44,8 +44,8 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.iceberg.ContentFile;
+import org.apache.impala.catalog.FeFsTable.FileMetadataStats;
import org.apache.impala.catalog.FeIcebergTable.Utils;
-import org.apache.impala.catalog.HdfsTable.FileMetadataStats;
import org.apache.impala.catalog.iceberg.GroupedContentFiles;
import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.common.Reference;
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
index 7a20d88bf..38d72914c 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
@@ -17,6 +17,7 @@
package org.apache.impala.catalog;
+import com.codahale.metrics.Gauge;
import com.codahale.metrics.Timer;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
@@ -237,6 +238,8 @@ public class IcebergTable extends Table implements FeIcebergTable {
private Map<Integer, IcebergColumn> icebergFieldIdToCol_;
private Map<String, TIcebergPartitionStats> partitionStats_;
+ private final FileMetadataStats fileMetadataStats_ = new FileMetadataStats();
+
protected IcebergTable(org.apache.hadoop.hive.metastore.api.Table msTable,
Db db, String name, String owner) {
super(msTable, db, name, owner);
@@ -405,6 +408,27 @@ public class IcebergTable extends Table implements FeIcebergTable {
return table;
}
+ @Override
+ public void initMetrics() {
+ super.initMetrics();
+ metrics_.addGauge(NUM_FILES_METRIC, new Gauge<Long>() {
+ @Override
+ public Long getValue() { return fileMetadataStats_.numFiles; }
+ });
+ metrics_.addGauge(NUM_BLOCKS_METRIC, new Gauge<Long>() {
+ @Override
+ public Long getValue() { return fileMetadataStats_.numBlocks; }
+ });
+ metrics_.addGauge(TOTAL_FILE_BYTES_METRIC, new Gauge<Long>() {
+ @Override
+ public Long getValue() { return fileMetadataStats_.totalFileBytes; }
+ });
+ metrics_.addGauge(MEMORY_ESTIMATE_METRIC, new Gauge<Long>() {
+ @Override
+ public Long getValue() { return getEstimatedMetadataSize(); }
+ });
+ }
+
/**
* Loads the metadata of an Iceberg table.
* <p>
@@ -461,19 +485,27 @@ public class IcebergTable extends Table implements FeIcebergTable {
loadAllColumnStats(msClient, catalogTimeline);
applyPuffinNdvStats(catalogTimeline);
setAvroSchema(msClient, msTbl, fileStore_, catalogTimeline);
+ updateMetrics(loader.getFileMetadataStats());
} catch (Exception e) {
throw new IcebergTableLoadingException("Error loading metadata for Iceberg table "
+ icebergTableLocation_, e);
} finally {
storageMetadataLoadTime_ = ctxStorageLdTime.stop();
}
-
refreshLastUsedTime();
} finally {
context.stop();
}
}
+ private void updateMetrics(FileMetadataStats stats) {
+ long memUsageEstimate = stats.numFiles * PER_FD_MEM_USAGE_BYTES +
+ stats.numBlocks * PER_BLOCK_MEM_USAGE_BYTES;
+ setEstimatedMetadataSize(memUsageEstimate);
+ setNumFiles(stats.numFiles);
+ fileMetadataStats_.set(stats);
+ }
+
// Reads NDV stats from Puffin files belonging to the table (if any).
//
// If there are Puffin stats for different snapshots, the most recent one will be used
diff --git a/tests/webserver/test_web_pages.py b/tests/webserver/test_web_pages.py
index a97d8b42b..52c8f59fe 100644
--- a/tests/webserver/test_web_pages.py
+++ b/tests/webserver/test_web_pages.py
@@ -21,6 +21,7 @@ from tests.common.file_utils import grep_dir
from tests.common.skip import SkipIfBuildType, SkipIfDockerizedCluster
from tests.common.impala_cluster import ImpalaCluster
from tests.common.impala_test_suite import ImpalaTestSuite
+from tests.util.filesystem_utils import IS_HDFS
from tests.util.parse_util import parse_duration_string_ms
from datetime import datetime
from multiprocessing import Process, Queue
@@ -1005,6 +1006,28 @@ class TestWebPage(ImpalaTestSuite):
assert "catalog.hms-client-pool.num-idle" in metric_keys
assert "catalog.hms-client-pool.num-in-use" in metric_keys
+ def test_iceberg_table_metrics(self):
+ assert '23448' == self.__get_table_metric(
+ "functional_parquet", "iceberg_non_partitioned", "total-file-size-bytes")
+ assert '20' == self.__get_table_metric(
+ "functional_parquet", "iceberg_non_partitioned", "num-files")
+ assert '13000' == self.__get_table_metric(
+ "functional_parquet", "iceberg_non_partitioned", "memory-estimate-bytes")
+ if IS_HDFS:
+ assert '20' == self.__get_table_metric(
+ "functional_parquet", "iceberg_non_partitioned", "num-blocks")
+
+ def __get_table_metric(self, db_name, tbl_name, metric):
+ self.client.execute("refresh %s.%s" % (db_name, tbl_name))
+ responses = self.get_and_check_status(self.TABLE_METRICS_URL + "?name=%s.%s&json" %
+ (db_name, tbl_name), metric, ports_to_test=self.CATALOG_TEST_PORT)
+ response_json = json.loads(responses[0].text)
+ metrics_text = response_json['table_metrics']
+ for line in metrics_text.split('\n'):
+ if line.startswith(metric):
+ return line.split(": ")[1]
+ return None
+
def test_query_progress(self):
"""Tests that /queries page shows query progress."""
query = "select count(*) from functional_parquet.alltypes where bool_col = sleep(100)"
(impala) 02/04: IMPALA-13789: Defer creating Path objects in loading file metadata
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 8c51f72e10388b0130811a9bfb594b51099b6bb6
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Sat Feb 22 20:00:56 2025 +0800
IMPALA-13789: Defer creating Path objects in loading file metadata
When loading file metadata of a HdfsTable, we create a
org.apache.hadoop.fs.Path object for each partition dir before actually
loading its file metadata. These Path objects have a large memory
footprint as the underlying java.net.URI objects have extra fields
extracted from the location string. E.g. a location string that takes
160 bytes has a corresponding Path object that takes 704 bytes. See more
details of this example in the JIRA description.
These Path objects are used as the keys of several Maps for file
metadata loaders. We create them before actually loading the metadata.
This patch fixes this by using the location strings as the keys and only
creating the Path objects when we start loading file metadata of the
partition.
Tests:
- Ran CORE tests
- Analyzed the heap dump during file metadata loading, didn't see lots
of Path objects anymore.
Change-Id: I6ec1fc932eaf7c833ef6ee6cdb08bba235e38271
Reviewed-on: http://gerrit.cloudera.org:8080/22535
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../apache/impala/catalog/CatalogHmsAPIHelper.java | 14 ++++-----
.../apache/impala/catalog/FileMetadataLoader.java | 34 ++++++++++++----------
.../impala/catalog/IcebergFileMetadataLoader.java | 28 ++++++++++--------
.../impala/catalog/ParallelFileMetadataLoader.java | 14 ++++-----
.../impala/catalog/local/DirectMetaProvider.java | 5 +---
.../org/apache/impala/common/FileSystemUtil.java | 15 ++++++++++
.../impala/catalog/FileMetadataLoaderTest.java | 16 +++++-----
.../apache/impala/catalog/HdfsPartitionTest.java | 2 +-
8 files changed, 71 insertions(+), 57 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogHmsAPIHelper.java b/fe/src/main/java/org/apache/impala/catalog/CatalogHmsAPIHelper.java
index ec5ebda6a..bb9140854 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogHmsAPIHelper.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogHmsAPIHelper.java
@@ -503,12 +503,11 @@ public class CatalogHmsAPIHelper {
checkCondition(tbl != null, "Table is null");
checkCondition(tbl.getSd() != null && tbl.getSd().getLocation() != null,
"Cannot get the location of table %s.%s", tbl.getDbName(), tbl.getTableName());
- Path tblPath = new Path(tbl.getSd().getLocation());
// since this table doesn't exist in catalogd we compute the network addresses
// for the files which are being returned.
ListMap<TNetworkAddress> hostIndex = new ListMap<>();
- FileMetadataLoader fmLoader = new FileMetadataLoader(tblPath, true,
- Collections.EMPTY_LIST, hostIndex, validTxnList, writeIdList);
+ FileMetadataLoader fmLoader = new FileMetadataLoader(tbl.getSd().getLocation(),
+ true, Collections.EMPTY_LIST, hostIndex, validTxnList, writeIdList);
boolean success = getFileMetadata(Arrays.asList(fmLoader));
checkCondition(success,
"Could not load file-metadata for table %s.%s. See catalogd log for details",
@@ -553,10 +552,9 @@ public class CatalogHmsAPIHelper {
checkCondition(part.getSd() != null && part.getSd().getLocation() != null,
"Could not get the location for partition %s of table %s.%s",
part.getValues(), part.getDbName(), part.getTableName());
- Path partPath = new Path(part.getSd().getLocation());
fileMdLoaders.put(part,
- new FileMetadataLoader(partPath, true, Collections.EMPTY_LIST, hostIndex,
- txnList, writeIdList));
+ new FileMetadataLoader(part.getSd().getLocation(), true,
+ Collections.EMPTY_LIST, hostIndex, txnList, writeIdList));
}
boolean success = getFileMetadata(fileMdLoaders.values());
checkCondition(success,
@@ -597,7 +595,7 @@ public class CatalogHmsAPIHelper {
* @return false if there were errors, else returns true.
*/
private static boolean getFileMetadata(Collection<FileMetadataLoader> loaders) {
- List<Pair<Path, Future<Void>>> futures = new ArrayList<>(loaders.size());
+ List<Pair<String, Future<Void>>> futures = new ArrayList<>(loaders.size());
for (FileMetadataLoader fmdLoader : loaders) {
futures.add(new Pair<>(fmdLoader.getPartDir(), fallbackFdLoaderPool.submit(() -> {
fmdLoader.load();
@@ -606,7 +604,7 @@ public class CatalogHmsAPIHelper {
}
int numberOfErrorsToLog = 100;
int errors = 0;
- for (Pair<Path, Future<Void>> pair : futures) {
+ for (Pair<String, Future<Void>> pair : futures) {
try {
pair.second.get();
} catch (InterruptedException | ExecutionException e) {
diff --git a/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
index 6298f5bac..214811241 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
@@ -59,7 +59,7 @@ public class FileMetadataLoader {
// at the end of load().
public static final AtomicInteger TOTAL_TASKS = new AtomicInteger();
- protected final Path partDir_;
+ protected final String partDir_;
protected final boolean recursive_;
protected final ImmutableMap<String, ? extends FileDescriptor> oldFdsByPath_;
private final ListMap<TNetworkAddress> hostIndex_;
@@ -96,7 +96,7 @@ public class FileMetadataLoader {
* @param fileFormat if non-null and equal to HdfsFileFormat.HUDI_PARQUET,
* this loader will filter files based on Hudi's HoodieROTablePathFilter method
*/
- public FileMetadataLoader(Path partDir, boolean recursive,
+ public FileMetadataLoader(String partDir, boolean recursive,
Iterable<? extends FileDescriptor> oldFds, ListMap<TNetworkAddress> hostIndex,
@Nullable ValidTxnList validTxnList, @Nullable ValidWriteIdList writeIds,
@Nullable HdfsFileFormat fileFormat) {
@@ -117,7 +117,7 @@ public class FileMetadataLoader {
TOTAL_TASKS.incrementAndGet();
}
- public FileMetadataLoader(Path partDir, boolean recursive,
+ public FileMetadataLoader(String partDir, boolean recursive,
Iterable<? extends FileDescriptor> oldFds, ListMap<TNetworkAddress> hostIndex,
@Nullable ValidTxnList validTxnList, @Nullable ValidWriteIdList writeIds) {
this(partDir, recursive, oldFds, hostIndex, validTxnList, writeIds, null);
@@ -158,7 +158,7 @@ public class FileMetadataLoader {
return loadStats_;
}
- Path getPartDir() { return partDir_; }
+ String getPartDir() { return partDir_; }
/**
* Load the file descriptors, which may later be fetched using {@link #getLoadedFds()}.
@@ -181,9 +181,10 @@ public class FileMetadataLoader {
private void loadInternal() throws CatalogException, IOException {
Preconditions.checkState(loadStats_ == null, "already loaded");
+ Path partPath = FileSystemUtil.createFullyQualifiedPath(new Path(partDir_));
loadStats_ = new LoadStats(partDir_);
fileMetadataStats_ = new FileMetadataStats();
- FileSystem fs = partDir_.getFileSystem(CONF);
+ FileSystem fs = partPath.getFileSystem(CONF);
// If we don't have any prior FDs from which we could re-use old block location info,
// we'll need to fetch info for every returned file. In this case we can inline
@@ -200,7 +201,7 @@ public class FileMetadataLoader {
listWithLocations ? " with eager location-fetching" : "", partDir_);
LOG.trace(msg);
try (ThreadNameAnnotator tna = new ThreadNameAnnotator(msg)) {
- List<FileStatus> fileStatuses = getFileStatuses(fs, listWithLocations);
+ List<FileStatus> fileStatuses = getFileStatuses(fs, partPath, listWithLocations);
loadedFds_ = new ArrayList<>();
if (fileStatuses == null) return;
@@ -208,7 +209,7 @@ public class FileMetadataLoader {
Reference<Long> numUnknownDiskIds = new Reference<>(0L);
if (writeIds_ != null) {
- fileStatuses = AcidUtils.filterFilesForAcidState(fileStatuses, partDir_,
+ fileStatuses = AcidUtils.filterFilesForAcidState(fileStatuses, partPath,
validTxnList_, writeIds_, loadStats_);
}
@@ -227,7 +228,7 @@ public class FileMetadataLoader {
}
FileDescriptor fd = getFileDescriptor(fs, listWithLocations, numUnknownDiskIds,
- fileStatus);
+ fileStatus, partPath);
loadedFds_.add(Preconditions.checkNotNull(fd));
fileMetadataStats_.accumulate(fd);
}
@@ -257,8 +258,9 @@ public class FileMetadataLoader {
* Return fd created by the given fileStatus or from the cache(oldFdsByPath_).
*/
protected FileDescriptor getFileDescriptor(FileSystem fs, boolean listWithLocations,
- Reference<Long> numUnknownDiskIds, FileStatus fileStatus) throws IOException {
- String relPath = FileSystemUtil.relativizePath(fileStatus.getPath(), partDir_);
+ Reference<Long> numUnknownDiskIds, FileStatus fileStatus, Path partPath)
+ throws IOException {
+ String relPath = FileSystemUtil.relativizePath(fileStatus.getPath(), partPath);
FileDescriptor fd = oldFdsByPath_.get(relPath);
if (listWithLocations || forceRefreshLocations || fd == null ||
fd.isChanged(fileStatus)) {
@@ -273,15 +275,15 @@ public class FileMetadataLoader {
/**
* Return located file status list when listWithLocations is true.
*/
- protected List<FileStatus> getFileStatuses(FileSystem fs, boolean listWithLocations)
- throws IOException {
+ protected List<FileStatus> getFileStatuses(FileSystem fs, Path partPath,
+ boolean listWithLocations) throws IOException {
RemoteIterator<? extends FileStatus> fileStatuses;
if (listWithLocations) {
fileStatuses = FileSystemUtil
- .listFiles(fs, partDir_, recursive_, debugAction_);
+ .listFiles(fs, partPath, recursive_, debugAction_);
} else {
fileStatuses = FileSystemUtil
- .listStatus(fs, partDir_, recursive_, debugAction_);
+ .listStatus(fs, partPath, recursive_, debugAction_);
// TODO(todd): we could look at the result of listing without locations, and if
// we see that a substantial number of the files have changed, it may be better
// to go back and re-list with locations vs doing an RPC per file.
@@ -349,8 +351,8 @@ public class FileMetadataLoader {
// File/Block metadata loading stats for a single HDFS path.
public static class LoadStats {
- private final Path partDir_;
- LoadStats(Path partDir) {
+ private final String partDir_;
+ LoadStats(String partDir) {
this.partDir_ = Preconditions.checkNotNull(partDir);
}
/** Number of files skipped because they pertain to an uncommitted ACID transaction */
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
index ac56f8d12..1b258467e 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
@@ -91,8 +91,8 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
Iterable<IcebergFileDescriptor> oldFds, ListMap<TNetworkAddress> hostIndex,
GroupedContentFiles icebergFiles, boolean requiresDataFilesInTableLocation,
int newFilesThresholdParam) {
- super(FileSystemUtil.createFullyQualifiedPath(new Path(iceTbl.location())), true,
- oldFds, hostIndex, null, null, HdfsFileFormat.ICEBERG);
+ super(iceTbl.location(), true, oldFds, hostIndex, null, null,
+ HdfsFileFormat.ICEBERG);
iceTbl_ = iceTbl;
icebergFiles_ = icebergFiles;
requiresDataFilesInTableLocation_ = requiresDataFilesInTableLocation;
@@ -125,17 +125,18 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
}
private void loadInternal() throws CatalogException, IOException {
+ Path partPath = FileSystemUtil.createFullyQualifiedPath(new Path(partDir_));
loadedFds_ = new ArrayList<>();
loadStats_ = new LoadStats(partDir_);
fileMetadataStats_ = new FileMetadataStats();
// Process the existing Fd ContentFile and return the newly added ContentFile
- Iterable<ContentFile<?>> newContentFiles = loadContentFilesWithOldFds();
+ Iterable<ContentFile<?>> newContentFiles = loadContentFilesWithOldFds(partPath);
// Iterate through all the newContentFiles, determine if StorageIds are supported,
// and use different handling methods accordingly.
// This considers that different ContentFiles are on different FileSystems
List<Pair<FileSystem, ContentFile<?>>> filesSupportsStorageIds = Lists.newArrayList();
- FileSystem fsForTable = FileSystemUtil.getFileSystemForPath(partDir_);
+ FileSystem fsForTable = FileSystemUtil.getFileSystemForPath(partPath);
FileSystem defaultFs = FileSystemUtil.getDefaultFileSystem();
for (ContentFile<?> contentFile : newContentFiles) {
FileSystem fsForPath = fsForTable;
@@ -151,7 +152,7 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
if (FileSystemUtil.supportsStorageIds(fsForPath)) {
filesSupportsStorageIds.add(Pair.create(fsForPath, contentFile));
} else {
- IcebergFileDescriptor fd = createFd(fsForPath, contentFile, null, null);
+ IcebergFileDescriptor fd = createFd(fsForPath, contentFile, null, partPath, null);
loadedFds_.add(fd);
fileMetadataStats_.accumulate(fd);
++loadStats_.loadedFiles;
@@ -172,7 +173,7 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
new Path(contentFileInfo.getSecond().path().toString()));
FileStatus stat = nameToFileStatus.get(path);
IcebergFileDescriptor fd = createFd(contentFileInfo.getFirst(),
- contentFileInfo.getSecond(), stat, numUnknownDiskIds);
+ contentFileInfo.getSecond(), stat, partPath, numUnknownDiskIds);
loadedFds_.add(fd);
fileMetadataStats_.accumulate(fd);
}
@@ -192,13 +193,14 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
* Iceberg tables are a collection of immutable, uniquely identifiable data files,
* which means we can safely reuse the old FDs.
*/
- private Iterable<ContentFile<?>> loadContentFilesWithOldFds() throws IOException {
+ private Iterable<ContentFile<?>> loadContentFilesWithOldFds(Path partPath)
+ throws IOException {
if (forceRefreshLocations || oldFdsByPath_.isEmpty()) {
return icebergFiles_.getAllContentFiles();
}
List<ContentFile<?>> newContentFiles = Lists.newArrayList();
for (ContentFile<?> contentFile : icebergFiles_.getAllContentFiles()) {
- FileDescriptor fd = getOldFd(contentFile);
+ FileDescriptor fd = getOldFd(contentFile, partPath);
if (fd == null) {
newContentFiles.add(contentFile);
} else {
@@ -211,7 +213,8 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
}
private IcebergFileDescriptor createFd(FileSystem fs, ContentFile<?> contentFile,
- FileStatus stat, Reference<Long> numUnknownDiskIds) throws IOException {
+ FileStatus stat, Path partPath, Reference<Long> numUnknownDiskIds)
+ throws IOException {
if (stat == null) {
Path fileLoc = FileSystemUtil.createFullyQualifiedPath(
new Path(contentFile.path().toString()));
@@ -224,7 +227,7 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
}
String absPath = null;
- String relPath = FileSystemUtil.relativizePathNoThrow(stat.getPath(), partDir_);
+ String relPath = FileSystemUtil.relativizePathNoThrow(stat.getPath(), partPath);
if (relPath == null) {
if (requiresDataFilesInTableLocation_) {
throw new IOException(String.format("Failed to load Iceberg datafile %s, because "
@@ -299,10 +302,11 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
return null;
}
- IcebergFileDescriptor getOldFd(ContentFile<?> contentFile) throws IOException {
+ IcebergFileDescriptor getOldFd(ContentFile<?> contentFile, Path partPath)
+ throws IOException {
Path contentFilePath = FileSystemUtil.createFullyQualifiedPath(
new Path(contentFile.path().toString()));
- String lookupPath = FileSystemUtil.relativizePathNoThrow(contentFilePath, partDir_);
+ String lookupPath = FileSystemUtil.relativizePathNoThrow(contentFilePath, partPath);
if (lookupPath == null) {
if (requiresDataFilesInTableLocation_) {
throw new IOException(String.format("Failed to load Iceberg datafile %s, because "
diff --git a/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
index 6f35f146b..385f3298a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
@@ -71,8 +71,8 @@ public class ParallelFileMetadataLoader {
private static final int MAX_PATH_METADATA_LOADING_ERRORS_TO_LOG = 100;
private final String logPrefix_;
- private final Map<Path, FileMetadataLoader> loaders_;
- private final Map<Path, List<HdfsPartition.Builder>> partsByPath_;
+ private final Map<String, FileMetadataLoader> loaders_;
+ private final Map<String, List<HdfsPartition.Builder>> partsByPath_;
private final FileSystem fs_;
public ParallelFileMetadataLoader(FileSystem fs,
@@ -89,13 +89,12 @@ public class ParallelFileMetadataLoader {
// path).
partsByPath_ = Maps.newHashMap();
for (HdfsPartition.Builder p : partBuilders) {
- Path partPath = FileSystemUtil.createFullyQualifiedPath(new Path(p.getLocation()));
- partsByPath_.computeIfAbsent(partPath, (path) -> new ArrayList<>())
+ partsByPath_.computeIfAbsent(p.getLocation(), (path) -> new ArrayList<>())
.add(p);
}
// Create a FileMetadataLoader for each path.
loaders_ = Maps.newHashMap();
- for (Map.Entry<Path, List<HdfsPartition.Builder>> e : partsByPath_.entrySet()) {
+ for (Map.Entry<String, List<HdfsPartition.Builder>> e : partsByPath_.entrySet()) {
List<FileDescriptor> oldFds = e.getValue().get(0).getFileDescriptors();
FileMetadataLoader loader;
HdfsFileFormat format = e.getValue().get(0).getFileFormat();
@@ -124,9 +123,8 @@ public class ParallelFileMetadataLoader {
loadInternal();
// Store the loaded FDs into the partitions.
- for (Map.Entry<Path, List<HdfsPartition.Builder>> e : partsByPath_.entrySet()) {
- Path p = e.getKey();
- FileMetadataLoader loader = loaders_.get(p);
+ for (Map.Entry<String, List<HdfsPartition.Builder>> e : partsByPath_.entrySet()) {
+ FileMetadataLoader loader = loaders_.get(e.getKey());
for (HdfsPartition.Builder partBuilder : e.getValue()) {
// Checks if we can reuse the old file descriptors. Partition builders in the list
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
index 5cc14a120..e52b98203 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
@@ -309,7 +309,6 @@ class DirectMetaProvider implements MetaProvider {
private Map<String, PartitionMetadata> loadUnpartitionedPartition(
TableMetaRefImpl table, List<PartitionRef> partitionRefs,
ListMap<TNetworkAddress> hostIndex) throws CatalogException {
- //TODO(IMPALA-9042): Remove "throws MetaException"
Preconditions.checkArgument(partitionRefs.size() == 1,
"Expected exactly one partition to load for unpartitioned table");
PartitionRef ref = partitionRefs.get(0);
@@ -381,13 +380,11 @@ class DirectMetaProvider implements MetaProvider {
private FileMetadataLoader loadFileMetadata(String fullTableName,
String partName, Partition msPartition, ListMap<TNetworkAddress> hostIndex)
throws CatalogException {
- //TODO(IMPALA-9042): Remove "throws MetaException"
- Path partDir = new Path(msPartition.getSd().getLocation());
// TODO(todd): The table property to disable recursive loading is not supported
// by this code path. However, DirectMetaProvider is not yet a supported feature.
// TODO(todd) this code path would have to change to handle ACID tables -- we don't
// have the write ID list passed down at this point in the code.
- FileMetadataLoader fml = new FileMetadataLoader(partDir,
+ FileMetadataLoader fml = new FileMetadataLoader(msPartition.getSd().getLocation(),
/* recursive= */BackendConfig.INSTANCE.recursivelyListPartitions(),
/* oldFds= */Collections.emptyList(),
hostIndex,
diff --git a/fe/src/main/java/org/apache/impala/common/FileSystemUtil.java b/fe/src/main/java/org/apache/impala/common/FileSystemUtil.java
index e17880db1..4ea9f6b78 100644
--- a/fe/src/main/java/org/apache/impala/common/FileSystemUtil.java
+++ b/fe/src/main/java/org/apache/impala/common/FileSystemUtil.java
@@ -736,6 +736,21 @@ public class FileSystemUtil {
return p.getFileSystem(CONF);
}
+ /**
+ * Returns the FileSystem object for a path string (might not be fully-qualified).
+ * I.e. a wrapper for getFileSystemForPath(createFullyQualifiedPath(new Path(p))).
+ */
+ public static FileSystem getFileSystemForPath(String p) throws IOException {
+ Path location = new Path(p);
+ URI defaultUri = FileSystem.getDefaultUri(CONF);
+ URI locationUri = location.toUri();
+ if (locationUri.getScheme() == null ||
+ locationUri.getScheme().equalsIgnoreCase(defaultUri.getScheme())) {
+ return getDefaultFileSystem();
+ }
+ return location.getFileSystem(CONF);
+ }
+
public static DistributedFileSystem getDistributedFileSystem() throws IOException {
FileSystem fs = getDefaultFileSystem();
Preconditions.checkState(fs instanceof DistributedFileSystem);
diff --git a/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java b/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
index a645c1081..bff749c34 100644
--- a/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
@@ -49,7 +49,7 @@ public class FileMetadataLoaderTest {
public void testRecursiveLoading() throws IOException, CatalogException {
//TODO(IMPALA-9042): Remove "throws CatalogException"
ListMap<TNetworkAddress> hostIndex = new ListMap<>();
- Path tablePath = new Path("hdfs://localhost:20500/test-warehouse/alltypes/");
+ String tablePath = "hdfs://localhost:20500/test-warehouse/alltypes/";
FileMetadataLoader fml = new FileMetadataLoader(tablePath, /* recursive=*/true,
/* oldFds = */Collections.emptyList(), hostIndex, null, null);
fml.load();
@@ -72,9 +72,9 @@ public class FileMetadataLoaderTest {
assertEquals(fml.getLoadedFds(), refreshFml.getLoadedFds());
// Touch a file and make sure that we reload locations for that file.
- FileSystem fs = tablePath.getFileSystem(new Configuration());
FileDescriptor fd = fml.getLoadedFds().get(0);
Path filePath = new Path(tablePath, fd.getRelativePath());
+ FileSystem fs = filePath.getFileSystem(new Configuration());
fs.setTimes(filePath, fd.getModificationTime() + 1, /* atime= */-1);
refreshFml = new FileMetadataLoader(tablePath, /* recursive=*/true,
@@ -87,7 +87,7 @@ public class FileMetadataLoaderTest {
public void testHudiParquetLoading() throws IOException, CatalogException {
//TODO(IMPALA-9042): Remove "throws CatalogException"
ListMap<TNetworkAddress> hostIndex = new ListMap<>();
- Path tablePath = new Path("hdfs://localhost:20500/test-warehouse/hudi_parquet/");
+ String tablePath = "hdfs://localhost:20500/test-warehouse/hudi_parquet/";
FileMetadataLoader fml = new FileMetadataLoader(tablePath, /* recursive=*/true,
/* oldFds = */ Collections.emptyList(), hostIndex, null, null,
HdfsFileFormat.HUDI_PARQUET);
@@ -358,8 +358,7 @@ public class FileMetadataLoaderTest {
ListMap<TNetworkAddress> hostIndex = new ListMap<>();
ValidWriteIdList writeIds =
MetastoreShim.getValidWriteIdListFromString(validWriteIdString);
- Path tablePath = new Path(path);
- FileMetadataLoader fml = new FileMetadataLoader(tablePath, /* recursive=*/true,
+ FileMetadataLoader fml = new FileMetadataLoader(path, /* recursive=*/true,
/* oldFds = */ Collections.emptyList(), hostIndex, new ValidReadTxnList(""),
writeIds, format);
fml.load();
@@ -393,7 +392,7 @@ public class FileMetadataLoaderTest {
//TODO(IMPALA-9042): Remove "throws CatalogException"
for (boolean recursive : ImmutableList.of(false, true)) {
ListMap<TNetworkAddress> hostIndex = new ListMap<>();
- Path tablePath = new Path("hdfs://localhost:20500/test-warehouse/does-not-exist/");
+ String tablePath = "hdfs://localhost:20500/test-warehouse/does-not-exist/";
FileMetadataLoader fml = new FileMetadataLoader(tablePath, recursive,
/* oldFds = */Collections.emptyList(), hostIndex, null, null);
fml.load();
@@ -405,7 +404,8 @@ public class FileMetadataLoaderTest {
@Test
public void testSkipHiddenDirectories() throws IOException, CatalogException {
Path sourcePath = new Path("hdfs://localhost:20500/test-warehouse/alltypes/");
- Path tmpTestPath = new Path("hdfs://localhost:20500/tmp/test-filemetadata-loader");
+ String tmpTestPathStr = "hdfs://localhost:20500/tmp/test-filemetadata-loader";
+ Path tmpTestPath = new Path(tmpTestPathStr);
Configuration conf = new Configuration();
FileSystem dstFs = tmpTestPath.getFileSystem(conf);
FileSystem srcFs = sourcePath.getFileSystem(conf);
@@ -421,7 +421,7 @@ public class FileMetadataLoaderTest {
dstFs.createNewFile(new Path(hiveStaging, "tmp-stats"));
dstFs.createNewFile(new Path(hiveStaging, ".hidden-tmp-stats"));
- FileMetadataLoader fml = new FileMetadataLoader(tmpTestPath, true,
+ FileMetadataLoader fml = new FileMetadataLoader(tmpTestPathStr, true,
Collections.emptyList(), new ListMap <>(), null, null);
fml.load();
assertEquals(24, fml.getStats().loadedFiles);
diff --git a/fe/src/test/java/org/apache/impala/catalog/HdfsPartitionTest.java b/fe/src/test/java/org/apache/impala/catalog/HdfsPartitionTest.java
index f4a0fb21d..ef2db6aec 100644
--- a/fe/src/test/java/org/apache/impala/catalog/HdfsPartitionTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/HdfsPartitionTest.java
@@ -147,7 +147,7 @@ public class HdfsPartitionTest {
@Test
public void testCloneWithNewHostIndex() throws Exception {
// Fetch some metadata from a directory in HDFS.
- Path p = new Path("hdfs://localhost:20500/test-warehouse/schemas");
+ String p = "hdfs://localhost:20500/test-warehouse/schemas";
ListMap<TNetworkAddress> origIndex = new ListMap<>();
FileMetadataLoader fml = new FileMetadataLoader(p, /* recursive= */false,
Collections.emptyList(), origIndex, /*validTxnList=*/null, /*writeIds=*/null);
(impala) 04/04: IMPALA-13804: Use redacted statement in live table
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 768527c89ad2ea3484fec0cd0bfdd56f54ab9046
Author: Michael Smith <mi...@cloudera.com>
AuthorDate: Wed Feb 26 14:41:58 2025 -0800
IMPALA-13804: Use redacted statement in live table
Uses the redacted SQL statement in sys.impala_query_live, so it's
consistent with the profile and sys.impala_query_log.
Testing: added a test with redaction rules for live and log tables.
Change-Id: I9a72eeaea84981e96655aec6c67b5ef2cbbd3c3e
Reviewed-on: http://gerrit.cloudera.org:8080/22556
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Jason Fehr <jf...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/system-table-scanner.cc | 2 +-
testdata/workload_mgmt/redaction.json | 11 +++++++++++
tests/custom_cluster/test_query_live.py | 16 +++++++++++++++-
tests/custom_cluster/test_query_log.py | 23 ++++++++++++++++++++++-
tests/util/workload_management.py | 7 +++++++
5 files changed, 56 insertions(+), 3 deletions(-)
diff --git a/be/src/exec/system-table-scanner.cc b/be/src/exec/system-table-scanner.cc
index e286c8353..229ad560c 100644
--- a/be/src/exec/system-table-scanner.cc
+++ b/be/src/exec/system-table-scanner.cc
@@ -363,7 +363,7 @@ Status QueryScanner::MaterializeNextTuple(
}
break;
case TQueryTableColumn::SQL:
- RETURN_IF_ERROR(WriteStringSlot(record.stmt, pool, slot));
+ RETURN_IF_ERROR(WriteStringSlot(query.redacted_sql, pool, slot));
break;
case TQueryTableColumn::PLAN:
RETURN_IF_ERROR(WriteStringSlot(
diff --git a/testdata/workload_mgmt/redaction.json b/testdata/workload_mgmt/redaction.json
new file mode 100644
index 000000000..3a086fcfe
--- /dev/null
+++ b/testdata/workload_mgmt/redaction.json
@@ -0,0 +1,11 @@
+{
+ "version": 1,
+ "rules": [
+ {
+ "description": "Don't show nonnsense",
+ "caseSensitive": false,
+ "search": "supercalifragilisticexpialidocious",
+ "replace": "*poppins*"
+ }
+ ]
+}
diff --git a/tests/custom_cluster/test_query_live.py b/tests/custom_cluster/test_query_live.py
index a0868d997..3175e544a 100644
--- a/tests/custom_cluster/test_query_live.py
+++ b/tests/custom_cluster/test_query_live.py
@@ -25,7 +25,7 @@ from signal import SIGRTMIN
from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
from tests.common.impala_cluster import DEFAULT_KRPC_PORT
from tests.util.retry import retry
-from tests.util.workload_management import assert_query
+from tests.util.workload_management import assert_query, redaction_rules_file
from time import sleep
@@ -225,6 +225,20 @@ class TestQueryLive(CustomClusterTestSuite):
assert_query('sys.impala_query_live', self.client, 'test_query_live',
result.runtime_profile)
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--cluster_id=test_query_live "
+ "--redaction_rules_file={}"
+ .format(redaction_rules_file()),
+ catalogd_args="--enable_workload_mgmt",
+ disable_log_buffering=True)
+ def test_redaction(self):
+ """Asserts the query live table table redacts the statement."""
+ result = self.client.execute(
+ "select *, 'supercalifragilisticexpialidocious' from functional.alltypes",
+ fetch_profile_after_close=True)
+ assert_query('sys.impala_query_live', self.client, 'test_query_live',
+ result.runtime_profile)
+
@CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
"--cluster_id=test_query_live",
catalogd_args="--enable_workload_mgmt",
diff --git a/tests/custom_cluster/test_query_log.py b/tests/custom_cluster/test_query_log.py
index 31dcf26c8..1d780e7ec 100644
--- a/tests/custom_cluster/test_query_log.py
+++ b/tests/custom_cluster/test_query_log.py
@@ -34,7 +34,11 @@ from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
from tests.common.impala_test_suite import IMPALAD_HS2_HOST_PORT
from tests.common.test_vector import ImpalaTestDimension
from tests.util.retry import retry
-from tests.util.workload_management import assert_query, WM_DB, QUERY_TBL_LOG
+from tests.util.workload_management import (
+ assert_query,
+ WM_DB,
+ QUERY_TBL_LOG,
+ redaction_rules_file)
class TestQueryLogTableBase(CustomClusterTestSuite):
@@ -410,6 +414,23 @@ class TestQueryLogTableBeeswax(TestQueryLogTableBase):
finally:
client2.close()
+ @CustomClusterTestSuite.with_args(impalad_args="--enable_workload_mgmt "
+ "--query_log_write_interval_s=1 "
+ "--redaction_rules_file={}"
+ .format(redaction_rules_file()),
+ catalogd_args="--enable_workload_mgmt",
+ disable_log_buffering=True)
+ def test_redaction(self):
+ """Asserts the query log table redacts the statement."""
+ result = self.client.execute(
+ "select *, 'supercalifragilisticexpialidocious' from functional.alltypes",
+ fetch_profile_after_close=True)
+ assert result.success
+
+ self.cluster.get_first_impalad().service.wait_for_metric_value(
+ "impala-server.completed-queries.written", 1, 60)
+ assert_query(QUERY_TBL_LOG, self.client, raw_profile=result.runtime_profile)
+
class TestQueryLogOtherTable(TestQueryLogTableBase):
"""Tests to assert that query_log_table_name works with non-default value."""
diff --git a/tests/util/workload_management.py b/tests/util/workload_management.py
index 9b2f567db..3a8a7c5b7 100644
--- a/tests/util/workload_management.py
+++ b/tests/util/workload_management.py
@@ -17,6 +17,7 @@
from __future__ import absolute_import, division, print_function
+import os
import re
import requests
@@ -733,3 +734,9 @@ def assert_csv_col(client, query_tbl, col, query_id, expected_list, db="tpcds"):
"'{}'\n actual (length {}): {}\n expected (length {}): {}" \
.format(TQueryTableColumn._VALUES_TO_NAMES[col], query_id, expected, len(actual),
sorted(actual), len(expected_list), sorted(expected_list))
+
+
+def redaction_rules_file():
+ """Provides the path to a redaction file that redacts the word
+ 'supercalifragilisticexpialidocious'."""
+ return "{}/testdata/workload_mgmt/redaction.json".format(os.environ["IMPALA_HOME"])
(impala) 01/04: IMPALA-13802: move postgresql init to the end of bin/bootstrap_system.sh
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
michaelsmith pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit a4f303ea55dbd299ee4153ca605851281269fe4c
Author: Csaba Ringhofer <cs...@cloudera.com>
AuthorDate: Wed Feb 26 17:01:23 2025 +0100
IMPALA-13802: move postgresql init to the end of bin/bootstrap_system.sh
postgresql initialization can fail if run a second time:
sudo service postgresql initdb
3 ERROR: Data directory /var/lib/pgsql/data is not empty!
This can led to skipping the rest of the script - a
quick fix is to deal with postgresql at the end.
Change-Id: I55e862ebe3b823e4aeaaa656d5536b6317b5e19c
Reviewed-on: http://gerrit.cloudera.org:8080/22550
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
bin/bootstrap_system.sh | 100 ++++++++++++++++++++++++++----------------------
1 file changed, 54 insertions(+), 46 deletions(-)
diff --git a/bin/bootstrap_system.sh b/bin/bootstrap_system.sh
index 9b45f31f8..2661138f8 100755
--- a/bin/bootstrap_system.sh
+++ b/bin/bootstrap_system.sh
@@ -396,52 +396,57 @@ fi
echo ">>> Configuring system"
-redhat notindocker sudo service postgresql initdb
-redhat notindocker sudo service postgresql stop
-redhat indocker sudo -u postgres PGDATA=/var/lib/pgsql/data pg_ctl init
-ubuntu sudo service postgresql stop
-
-# These configurations expose connectiong to PostgreSQL via md5-hashed
-# passwords over TCP to localhost, and the local socket is trusted
-# widely.
-ubuntu sudo sed -ri 's/local +all +all +peer/local all all trust/g' \
- /etc/postgresql/*/main/pg_hba.conf
-# Accept remote connections from the hosts in the same subnet.
-ubuntu sudo sed -ri "s/#listen_addresses = 'localhost'/listen_addresses = '0.0.0.0'/g" \
- /etc/postgresql/*/main/postgresql.conf
-ubuntu sudo sed -ri 's/host +all +all +127.0.0.1\/32/host all all samenet/g' \
- /etc/postgresql/*/main/pg_hba.conf
-redhat sudo sed -ri 's/local +all +all +(ident|peer)/local all all trust/g' \
- /var/lib/pgsql/data/pg_hba.conf
-# Accept md5 passwords from localhost
-redhat sudo sed -i -e 's,\(host.*\)ident,\1md5,' /var/lib/pgsql/data/pg_hba.conf
-# Accept remote connections from the hosts in the same subnet.
-redhat sudo sed -ri "s/#listen_addresses = 'localhost'/listen_addresses = '0.0.0.0'/g" \
- /var/lib/pgsql/data/postgresql.conf
-redhat sudo sed -ri 's/host +all +all +127.0.0.1\/32/host all all samenet/g' \
- /var/lib/pgsql/data/pg_hba.conf
-
-ubuntu sudo service postgresql start
-redhat notindocker sudo service postgresql start
-# Important to redirect pg_ctl to a logfile, lest it keep the stdout
-# file descriptor open, preventing the shell from exiting.
-redhat indocker sudo -u postgres PGDATA=/var/lib/pgsql/data bash -c \
- "pg_ctl start -w --timeout=120 >> /var/lib/pgsql/pg.log 2>&1"
-
-# Set up postgres for HMS
-if ! [[ 1 = $(sudo -u postgres psql -At -c "SELECT count(*) FROM pg_roles WHERE rolname = 'hiveuser';") ]]
-then
- sudo -u postgres psql -c "CREATE ROLE hiveuser LOGIN PASSWORD 'password';"
-fi
-sudo -u postgres psql -c "ALTER ROLE hiveuser WITH CREATEDB;"
-# On Ubuntu 18.04 aarch64 version, the sql 'select * from pg_roles' blocked,
-# because output of 'select *' is too long to display in 1 line.
-# So here just change it to 'select count(*)' as a work around.
-if [[ $ARCH_NAME == 'aarch64' ]]; then
- sudo -u postgres psql -c "SELECT count(*) FROM pg_roles WHERE rolname = 'hiveuser';"
-else
- sudo -u postgres psql -c "SELECT * FROM pg_roles WHERE rolname = 'hiveuser';"
-fi
+function setup_postgresql() {
+ echo ">>> Configuring postgresql. This can fail if postgres is already initialized"
+
+ redhat notindocker sudo service postgresql initdb
+ redhat notindocker sudo service postgresql stop
+ redhat indocker sudo -u postgres PGDATA=/var/lib/pgsql/data pg_ctl init
+ ubuntu sudo service postgresql stop
+
+ # These configurations expose connectiong to PostgreSQL via md5-hashed
+ # passwords over TCP to localhost, and the local socket is trusted
+ # widely.
+ ubuntu sudo sed -ri 's/local +all +all +peer/local all all trust/g' \
+ /etc/postgresql/*/main/pg_hba.conf
+ # Accept remote connections from the hosts in the same subnet.
+ ubuntu sudo sed -ri "s/#listen_addresses = 'localhost'/listen_addresses = '0.0.0.0'/g" \
+ /etc/postgresql/*/main/postgresql.conf
+ ubuntu sudo sed -ri 's/host +all +all +127.0.0.1\/32/host all all samenet/g' \
+ /etc/postgresql/*/main/pg_hba.conf
+ redhat sudo sed -ri 's/local +all +all +(ident|peer)/local all all trust/g' \
+ /var/lib/pgsql/data/pg_hba.conf
+ # Accept md5 passwords from localhost
+ redhat sudo sed -i -e 's,\(host.*\)ident,\1md5,' /var/lib/pgsql/data/pg_hba.conf
+ # Accept remote connections from the hosts in the same subnet.
+ redhat sudo sed -ri "s/#listen_addresses = 'localhost'/listen_addresses = '0.0.0.0'/g" \
+ /var/lib/pgsql/data/postgresql.conf
+ redhat sudo sed -ri 's/host +all +all +127.0.0.1\/32/host all all samenet/g' \
+ /var/lib/pgsql/data/pg_hba.conf
+
+ ubuntu sudo service postgresql start
+ redhat notindocker sudo service postgresql start
+ # Important to redirect pg_ctl to a logfile, lest it keep the stdout
+ # file descriptor open, preventing the shell from exiting.
+ redhat indocker sudo -u postgres PGDATA=/var/lib/pgsql/data bash -c \
+ "pg_ctl start -w --timeout=120 >> /var/lib/pgsql/pg.log 2>&1"
+
+ # Set up postgres for HMS
+ if ! [[ 1 = $(sudo -u postgres psql -At -c "SELECT count(*) FROM pg_roles WHERE rolname = 'hiveuser';") ]]
+ then
+ sudo -u postgres psql -c "CREATE ROLE hiveuser LOGIN PASSWORD 'password';"
+ fi
+ sudo -u postgres psql -c "ALTER ROLE hiveuser WITH CREATEDB;"
+ # On Ubuntu 18.04 aarch64 version, the sql 'select * from pg_roles' blocked,
+ # because output of 'select *' is too long to display in 1 line.
+ # So here just change it to 'select count(*)' as a work around.
+ if [[ $ARCH_NAME == 'aarch64' ]]; then
+ sudo -u postgres psql -c "SELECT count(*) FROM pg_roles WHERE rolname = 'hiveuser';"
+ else
+ sudo -u postgres psql -c "SELECT * FROM pg_roles WHERE rolname = 'hiveuser';"
+ fi
+ echo ">>> Configuring postgresql finished."
+}
# Setup ssh to ssh to localhost
mkdir -p ~/.ssh
@@ -527,3 +532,6 @@ if [[ "${PREPOPULATE_M2_REPOSITORY:-true}" == true ]] ; then
else
echo ">>> Skip populating m2 directory"
fi
+
+setup_postgresql
+# Be careful about adding code after postgres initialization, as it may fail and exit!