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 2026/08/05 17:08:48 UTC
(impala) branch master updated (664f455ad -> 538ceb422)
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 664f455ad IMPALA-15174: Add AGGREGATED_PROFILE query option to toggle profile formats
new 6c32a71ef IMPALA-15171: Null out Iceberg delete file path slot after the join
new 538ceb422 IMPALA-14684: Add metrics for catalogd table invalidations
The 2 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/catalog/catalog-server.cc | 85 +++++++++
be/src/catalog/catalog-server.h | 25 +++
be/src/exec/file-metadata-utils.cc | 8 +
be/src/exec/hdfs-scan-node-base.cc | 3 +
be/src/exec/hdfs-scan-node-base.h | 8 +
be/src/exec/iceberg-delete-node.cc | 17 ++
be/src/exec/iceberg-delete-node.h | 14 ++
common/thrift/JniCatalog.thrift | 24 +++
common/thrift/PlanNodes.thrift | 15 ++
common/thrift/metrics.json | 140 ++++++++++++++
.../java/org/apache/impala/catalog/Catalog.java | 40 +++-
.../impala/catalog/CatalogServiceCatalog.java | 2 +-
.../impala/catalog/CatalogdTableInvalidator.java | 172 ++++++++++++++++++
.../org/apache/impala/planner/HdfsScanNode.java | 5 +-
.../impala/planner/IcebergDeleteJoinNode.java | 12 +-
.../org/apache/impala/planner/IcebergScanNode.java | 31 +++-
.../apache/impala/planner/IcebergScanPlanner.java | 41 ++++-
.../java/org/apache/impala/service/JniCatalog.java | 32 ++++
.../catalog/CatalogdTableInvalidatorTest.java | 87 +++++++++
.../queries/PlannerTest/iceberg-optimize.test | 2 +
.../PlannerTest/iceberg-v2-tables-resources.test | 66 +++++++
.../queries/PlannerTest/tablesample-iceberg.test | 4 +
.../custom_cluster/test_automatic_invalidation.py | 201 +++++++++++++++++++++
tests/query_test/test_iceberg.py | 59 +++++-
24 files changed, 1074 insertions(+), 19 deletions(-)
(impala) 01/02: IMPALA-15171: Null out Iceberg delete file path slot after the join
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 6c32a71ef2ffb4e8fbf4f6ab10ad95fe01b6b875
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
AuthorDate: Fri Jul 10 18:03:20 2026 +0200
IMPALA-15171: Null out Iceberg delete file path slot after the join
IcebergScanPlanner materializes the INPUT__FILE__NAME (file path)
virtual slot on the data scan tuple so the position-delete join
(IcebergDeleteJoinNode) can use it as a join key. When the query does
not otherwise need the file path, the slot is still carried in the
join's output tuple and propagates through downstream EXCHANGE
operators, where the (typically long) file path string is deep-copied
and serialized for every surviving row.
The planner now detects when it materialized the file path slot solely
for the delete join, i.e. no operator above the join references it. When
the flag is set, IcebergDeleteNode nulls out the file path slot on
its output rows in GetNext() so it won't get copied downstream.
When some data files have deletes and some do not, the plan is a UNION
ALL of the delete anti-join (data files with deletes) and a plain scan
(data files without deletes). Both branches share the same tuple
descriptor, so the without-deletes scan also materializes the file path
slot even though it never goes through IcebergDeleteNode. Previously
nothing nulled that branch, so it still propagated the file path
string up through the UNION.
This is now handled symmetrically: the same flag is passed to the
without-deletes IcebergScanNode via THdfsScanNode.clear_file_path_slot,
and FileMetadataUtils::AddFileLevelVirtualColumns leaves the
INPUT__FILE__NAME slot NULL instead of assigning the filename. The tuple
layout is unchanged, so UNION pass-through still holds. The with-deletes
data scan is unaffected (it still needs the path as the join key; the
delete node nulls it after the lookup).
A new tag "clear file path slot" is added to operators (printed with
EXTENDED level or above) that are configured to clear the file path
slot.
Improvement
I used the following query (ice_lineitem has data files with and without
deletes):
SELECT count(*)
FROM ice_lineitem a, ice_lineitem b
WHERE a.l_orderkey=b.l_orderkey;
TotalBytesSent: 914.26 MB -> 246.84 MB
HASH JOIN Peak mem usage: 558.05 MB -> 238.05 MB
Testing
* added e2e test for the above query
* updated planner tests
Change-Id: I7e662cb6a98dd3e687185d384731d6be45f91b94
Assisted-by: Claude Opus 4.8 (1M context) <no...@anthropic.com>
Reviewed-on: http://gerrit.cloudera.org:8080/24572
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/file-metadata-utils.cc | 8 +++
be/src/exec/hdfs-scan-node-base.cc | 3 +
be/src/exec/hdfs-scan-node-base.h | 8 +++
be/src/exec/iceberg-delete-node.cc | 17 ++++++
be/src/exec/iceberg-delete-node.h | 14 +++++
common/thrift/PlanNodes.thrift | 15 +++++
.../org/apache/impala/planner/HdfsScanNode.java | 5 +-
.../impala/planner/IcebergDeleteJoinNode.java | 12 +++-
.../org/apache/impala/planner/IcebergScanNode.java | 31 ++++++++--
.../apache/impala/planner/IcebergScanPlanner.java | 41 ++++++++++++--
.../queries/PlannerTest/iceberg-optimize.test | 2 +
.../PlannerTest/iceberg-v2-tables-resources.test | 66 ++++++++++++++++++++++
.../queries/PlannerTest/tablesample-iceberg.test | 4 ++
tests/query_test/test_iceberg.py | 59 ++++++++++++++++++-
14 files changed, 268 insertions(+), 17 deletions(-)
diff --git a/be/src/exec/file-metadata-utils.cc b/be/src/exec/file-metadata-utils.cc
index 45df6db4e..5a7d63c94 100644
--- a/be/src/exec/file-metadata-utils.cc
+++ b/be/src/exec/file-metadata-utils.cc
@@ -105,6 +105,14 @@ void FileMetadataUtils::AddFileLevelVirtualColumns(MemPool* mem_pool,
for (int i = 0; i < scan_node_->virtual_column_slots().size(); ++i) {
const SlotDescriptor* slot_desc = scan_node_->virtual_column_slots()[i];
if (slot_desc->virtual_column_type() == TVirtualColumnType::INPUT_FILE_NAME) {
+ if (scan_node_->clear_file_path_slot()) {
+ // The INPUT__FILE__NAME slot was materialized only as a join key for the
+ // position-delete join on the sibling (with-deletes) UNION branch. Nothing above
+ // references it, so leave it NULL here to avoid propagating the (typically long)
+ // file path string. Mirrors IcebergDeleteNode NULL-ing its join output.
+ template_tuple->SetNull(slot_desc->null_indicator_offset());
+ continue;
+ }
StringValue* slot = template_tuple->GetStringSlot(slot_desc->tuple_offset());
const char* filename = file_desc_->filename.c_str();
int len = strlen(filename);
diff --git a/be/src/exec/hdfs-scan-node-base.cc b/be/src/exec/hdfs-scan-node-base.cc
index 961faa501..fbe392649 100644
--- a/be/src/exec/hdfs-scan-node-base.cc
+++ b/be/src/exec/hdfs-scan-node-base.cc
@@ -474,6 +474,9 @@ HdfsScanNodeBase::HdfsScanNodeBase(ObjectPool* pool, const HdfsScanPlanNode& pno
hdfs_scan_node.count_star_slot_offset :
-1),
is_partition_key_scan_(hdfs_scan_node.is_partition_key_scan),
+ clear_file_path_slot_(hdfs_scan_node.__isset.clear_file_path_slot ?
+ hdfs_scan_node.clear_file_path_slot :
+ false),
tuple_desc_(pnode.tuple_desc_),
hdfs_table_(pnode.hdfs_table_),
avro_schema_(*pnode.avro_schema_.get()),
diff --git a/be/src/exec/hdfs-scan-node-base.h b/be/src/exec/hdfs-scan-node-base.h
index 5df535d05..5e8a4bbab 100644
--- a/be/src/exec/hdfs-scan-node-base.h
+++ b/be/src/exec/hdfs-scan-node-base.h
@@ -470,6 +470,7 @@ class HdfsScanNodeBase : public ScanNode {
bool optimize_count_star() const { return count_star_slot_offset_ != -1; }
int count_star_slot_offset() const { return count_star_slot_offset_; }
bool is_partition_key_scan() const { return is_partition_key_scan_; }
+ bool clear_file_path_slot() const { return clear_file_path_slot_; }
typedef std::unordered_map<TupleId, std::vector<ScalarExprEvaluator*>>
ConjunctEvaluatorsMap;
@@ -699,6 +700,13 @@ class HdfsScanNodeBase : public ScanNode {
// to do the minimum possible work to materialise one row.
const bool is_partition_key_scan_;
+ /// True if the INPUT__FILE__NAME virtual slot on this scan's tuple was materialized
+ /// only as a join key for a sibling IcebergDeleteNode (UNION ALL branch) and nothing
+ /// above references it. When true, the scanner leaves the INPUT__FILE__NAME slot NULL
+ /// instead of assigning the file path, so the (typically long) path string is not
+ /// propagated unnecessarily. See IMPALA-15171.
+ const bool clear_file_path_slot_;
+
/// RequestContext object to use with the disk-io-mgr for reads.
std::unique_ptr<io::RequestContext> reader_context_;
diff --git a/be/src/exec/iceberg-delete-node.cc b/be/src/exec/iceberg-delete-node.cc
index 16e3ce551..5392a4bc0 100644
--- a/be/src/exec/iceberg-delete-node.cc
+++ b/be/src/exec/iceberg-delete-node.cc
@@ -40,6 +40,8 @@ Status IcebergDeletePlanNode::Init(const TPlanNode& tnode, FragmentState* state)
DCHECK(tnode.__isset.join_node);
DCHECK(tnode.join_node.__isset.iceberg_delete_node);
+ clear_file_path_slot_ = tnode.join_node.iceberg_delete_node.clear_file_path_slot;
+
// TODO: IMPALA-12265: create the config only if it is necessary
RETURN_IF_ERROR(IcebergDeleteBuilderConfig::CreateConfig(state, tnode_->node_id,
tnode_->join_node.join_op, &build_row_desc(), &id_builder_config_));
@@ -78,6 +80,9 @@ Status IcebergDeleteNode::Prepare(RuntimeState* state) {
runtime_profile()->PrependChild(builder_->profile());
}
+ clear_file_path_slot_ =
+ static_cast<const IcebergDeletePlanNode&>(plan_node_).clear_file_path_slot_;
+
auto& tuple_descs = probe_row_desc().tuple_descriptors();
auto& slot_descs = tuple_descs[0]->slots();
@@ -87,6 +92,7 @@ Status IcebergDeleteNode::Prepare(RuntimeState* state) {
}
if (slot->virtual_column_type() == TVirtualColumnType::INPUT_FILE_NAME) {
file_path_offset_ = slot->tuple_offset();
+ file_path_null_indicator_offset_ = slot->null_indicator_offset();
}
}
@@ -260,6 +266,17 @@ Status IcebergDeleteNode::GetNext(RuntimeState* state, RowBatch* out_batch, bool
*eos = true;
}
+ // The file path slot is only needed as a join key by this node. If nothing above
+ // references it, NULL it out on the rows produced by this call so the (typically long)
+ // file path string is not deep-copied and serialized through downstream exchanges.
+ // Output rows share the probe tuples (CopyRows() shallow-copies tuple pointers), and
+ // those tuples are never read again once produced, so this is safe.
+ if (clear_file_path_slot_) {
+ for (int i = num_rows_before; i < num_rows_before + num_rows_added; ++i) {
+ out_batch->GetRow(i)->GetTuple(0)->SetNull(file_path_null_indicator_offset_);
+ }
+ }
+
IncrementNumRowsReturned(num_rows_added);
COUNTER_SET(rows_returned_counter_, rows_returned());
return Status::OK();
diff --git a/be/src/exec/iceberg-delete-node.h b/be/src/exec/iceberg-delete-node.h
index 0355a7f3f..b605dde59 100644
--- a/be/src/exec/iceberg-delete-node.h
+++ b/be/src/exec/iceberg-delete-node.h
@@ -41,6 +41,12 @@ class IcebergDeletePlanNode : public BlockingJoinPlanNode {
/// Data sink config object for creating a id builder that will be eventually used by
/// the exec node.
IcebergDeleteBuilderConfig* id_builder_config_;
+
+ /// If true, the exec node nulls out the INPUT__FILE__NAME (file path) slot on output
+ /// rows. The file path is only needed as a join key here; nulling it avoids
+ /// deep-copying and serializing the (typically long) file path string through
+ /// downstream exchanges. See IMPALA-15171.
+ bool clear_file_path_slot_ = false;
};
/// Operator to perform iceberg delete.
@@ -179,6 +185,14 @@ class IcebergDeleteNode : public BlockingJoinNode {
int file_path_offset_;
int pos_offset_;
+
+ /// Null indicator of the INPUT__FILE__NAME (file path) slot in the probe tuple, used
+ /// to NULL out the slot on output rows when 'clear_file_path_slot_' is true.
+ NullIndicatorOffset file_path_null_indicator_offset_{0, -1};
+
+ /// Cached from IcebergDeletePlanNode::clear_file_path_slot_ in Prepare(). When true,
+ /// the file path slot is nulled on output rows. See IMPALA-15171.
+ bool clear_file_path_slot_ = false;
};
} // namespace impala
diff --git a/common/thrift/PlanNodes.thrift b/common/thrift/PlanNodes.thrift
index 62a69e9e4..b9e321169 100644
--- a/common/thrift/PlanNodes.thrift
+++ b/common/thrift/PlanNodes.thrift
@@ -379,6 +379,14 @@ struct THdfsScanNode {
// when new files are being added to a table. For tuple caching, this can improve the
// cache hits by avoiding unnecessary disruption to the runtime keys.
16: optional bool schedule_scanranges_oldest_to_newest
+
+ // If true, this Iceberg data scan reads data files that have no delete files and is a
+ // sibling (UNION ALL branch) of an IcebergDeleteNode. The INPUT__FILE__NAME virtual
+ // slot was materialized on the shared tuple solely as a join key for that delete node,
+ // and nothing above references it. The scanner leaves the slot NULL so the (typically
+ // long) file path string is not propagated/serialized unnecessarily. Mirrors
+ // TIcebergDeleteNode.clear_file_path_slot on the with-deletes branch. See IMPALA-15171.
+ 17: optional bool clear_file_path_slot
}
struct TDataSourceScanNode {
@@ -527,6 +535,13 @@ struct TNestedLoopJoinNode {
struct TIcebergDeleteNode {
// equi-join predicates
1: required list<TEqJoinCondition> eq_join_conjuncts
+
+ // If true, the backend nulls out the INPUT__FILE__NAME (file path) slot on the join's
+ // output rows. The file path is only needed as a join key at this node; nulling it
+ // avoids deep-copying and serializing the (typically long) file path string through
+ // downstream exchanges. Only set by the planner when it materialized the file path
+ // slot itself, i.e. no operator above the join references it. See IMPALA-15171.
+ 2: optional bool clear_file_path_slot
}
// Top-level struct for a join node. Elements that are shared between the different
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
index 3ca54c7de..34e155885 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
@@ -2085,10 +2085,9 @@ public class HdfsScanNode extends ScanNode {
if (isPartitionKeyScan_) {
output.append(detailPrefix + "partition key scan\n");
}
-
- String derivedExplain = getDerivedExplainString(detailPrefix, detailLevel);
- if (!derivedExplain.isEmpty()) output.append(derivedExplain);
}
+ String derivedExplain = getDerivedExplainString(detailPrefix, detailLevel);
+ if (!derivedExplain.isEmpty()) output.append(derivedExplain);
if (detailLevel.ordinal() >= TExplainLevel.EXTENDED.ordinal()) {
output.append(getStatsExplainString(detailPrefix)).append("\n");
String extrapRows;
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergDeleteJoinNode.java b/fe/src/main/java/org/apache/impala/planner/IcebergDeleteJoinNode.java
index 91ecaffad..86d7660dd 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergDeleteJoinNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergDeleteJoinNode.java
@@ -50,12 +50,18 @@ public class IcebergDeleteJoinNode extends JoinNode {
// Record count from deletion vectors associated with the scanned data files.
private final long dvDeletesRecordCount_;
+ // If true, the backend nulls out the INPUT__FILE__NAME (file path) slot on the output
+ // rows. Only set when the planner materialized the file path slot solely for this
+ // join, i.e. no operator above references it. See IMPALA-15171.
+ private final boolean clearFilePathSlot_;
+
public IcebergDeleteJoinNode(
PlanNode outer, PlanNode inner, List<BinaryPredicate> eqJoinConjuncts,
- long dvDeletesRecordCount) {
+ long dvDeletesRecordCount, boolean clearFilePathSlot) {
super(outer, inner, true, DistributionMode.NONE, JoinOperator.ICEBERG_DELETE_JOIN,
eqJoinConjuncts, Collections.emptyList(), "ICEBERG DELETE");
dvDeletesRecordCount_ = dvDeletesRecordCount;
+ clearFilePathSlot_ = clearFilePathSlot;
Preconditions.checkNotNull(eqJoinConjuncts);
Preconditions.checkState(joinOp_ == JoinOperator.ICEBERG_DELETE_JOIN);
Preconditions.checkState(conjuncts_.isEmpty());
@@ -147,6 +153,7 @@ public class IcebergDeleteJoinNode extends JoinNode {
msg.join_node.iceberg_delete_node = new TIcebergDeleteNode();
msg.join_node.iceberg_delete_node.setEq_join_conjuncts(
getThriftEquiJoinConjuncts(serialCtx));
+ msg.join_node.iceberg_delete_node.setClear_file_path_slot(clearFilePathSlot_);
}
@Override
@@ -168,6 +175,9 @@ public class IcebergDeleteJoinNode extends JoinNode {
output.append("\n");
}
}
+ if (detailLevel.ordinal() >= TExplainLevel.EXTENDED.ordinal()) {
+ if (clearFilePathSlot_) output.append(detailPrefix + "clear file path slot\n");
+ }
return output.toString();
}
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
index f96704214..8743e880b 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanNode.java
@@ -98,6 +98,12 @@ public class IcebergScanNode extends HdfsScanNode {
// responsible for reading the delete files of the corresponding table.
private final PlanNodeId deleteFileScanNodeId;
+ // When true, the INPUT__FILE__NAME virtual slot is materialized on this scan's tuple
+ // only as a join key for a sibling IcebergDeleteNode (UNION ALL branch), and nothing
+ // above references it. The backend scanner leaves the slot NULL so the file path string
+ // is not propagated unnecessarily. See IMPALA-15171.
+ private boolean clearFilePathSlot_ = false;
+
public IcebergScanNode(PlanNodeId id, TableRef tblRef, List<Expr> conjuncts,
MultiAggregateInfo aggInfo, List<IcebergFileDescriptor> fileDescs,
int numPartitions,
@@ -248,6 +254,14 @@ public class IcebergScanNode extends HdfsScanNode {
if (deleteFileScanNodeId != null) {
msg.hdfs_scan_node.setDeleteFileScanNodeId(deleteFileScanNodeId.asInt());
}
+ if (clearFilePathSlot_) {
+ msg.hdfs_scan_node.setClear_file_path_slot(true);
+ }
+ }
+
+ // See clearFilePathSlot_.
+ public void setClearFilePathSlot(boolean clearFilePathSlot) {
+ clearFilePathSlot_ = clearFilePathSlot;
}
@Override
@@ -287,12 +301,17 @@ public class IcebergScanNode extends HdfsScanNode {
protected String getDerivedExplainString(
String indentPrefix, TExplainLevel detailLevel) {
StringBuilder output = new StringBuilder();
- output.append(
- indentPrefix + "Iceberg snapshot id: " + String.valueOf(snapshotId_) + "\n");
- if (!skippedConjuncts_.isEmpty()) {
- output.append(indentPrefix +
- String.format("skipped Iceberg predicates: %s\n",
- Expr.getExplainString(skippedConjuncts_, detailLevel)));
+ if (detailLevel.ordinal() >= TExplainLevel.STANDARD.ordinal()) {
+ output.append(
+ indentPrefix + "Iceberg snapshot id: " + String.valueOf(snapshotId_) + "\n");
+ if (!skippedConjuncts_.isEmpty()) {
+ output.append(indentPrefix +
+ String.format("skipped Iceberg predicates: %s\n",
+ Expr.getExplainString(skippedConjuncts_, detailLevel)));
+ }
+ }
+ if (detailLevel.ordinal() >= TExplainLevel.EXTENDED.ordinal()) {
+ if (clearFilePathSlot_) output.append(indentPrefix + "clear file path slot\n");
}
return output.toString();
}
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java b/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
index 56275d87c..c2c7fbb86 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
@@ -160,6 +160,12 @@ public class IcebergScanPlanner {
private long dvDeletesRecordCount_ = 0;
private long dataFilesWithDeletesSumPaths_ = 0;
private long dataFilesWithDeletesMaxPath_ = 0;
+ // True if the planner materialized the INPUT__FILE__NAME (file path) slot solely as a
+ // join key for the position-delete ANTI JOIN, i.e. nothing above the scan references
+ // it. When true, the file path slot is NULLed out both by the IcebergDeleteNode (on the
+ // with-deletes branch) and by the data scan node on the without-deletes UNION branch.
+ // Set in createPositionJoinNode(). See IMPALA-15171.
+ private boolean clearFilePathSlot_ = false;
// Stores how many delete records are involved broken down by equality field ID lists.
private Map<List<Integer>, Long> equalityDeletesRecordCount_ = new HashMap<>();
private Set<Long> equalityDeleteSequenceNumbers_ = new HashSet<>();
@@ -308,6 +314,11 @@ public class IcebergScanPlanner {
getIceTable().getContentFileStore().getNumPartitions(),
nonIdentityConjuncts_, getSkippedConjuncts(), snapshotId_,
isPartitionKeyScan, dataFileToDV_, helper_);
+ // These data files have no deletes, so they don't go through the IcebergDeleteNode
+ // that NULLs out the file path slot. If that slot was materialized only as a join key
+ // (clearFilePathSlot_), NULL it out here too so the file path string isn't propagated
+ // up through the UNION unnecessarily.
+ dataScanNode.setClearFilePathSlot(clearFilePathSlot_);
dataScanNode.init(analyzer_);
List<Expr> outputExprs = tblRef_.getDesc().getSlots().stream().map(
SlotRef::new).collect(Collectors.toList());
@@ -360,7 +371,7 @@ public class IcebergScanPlanner {
TableRef deleteDeltaRef = TableRef.newTableRef(analyzer_,
Arrays.asList(deleteTable.getDb().getName(), deleteTable.getName()),
tblRef_.getUniqueAlias() + "-position-delete");
- addDataVirtualPositionSlots(tblRef_);
+ clearFilePathSlot_ = addDataVirtualPositionSlots(tblRef_);
if (!equalityIdsToDeleteFiles_.isEmpty()) addAllSlotsForEqualityDeletes(tblRef_);
addDeletePositionSlots(deleteDeltaRef);
IcebergScanNode dataScanNode = new IcebergScanNode(
@@ -396,7 +407,7 @@ public class IcebergScanPlanner {
} else {
joinNode =
new IcebergDeleteJoinNode(dataScanNode, deleteScanNode, positionJoinConjuncts,
- dvDeletesRecordCount_);
+ dvDeletesRecordCount_, clearFilePathSlot_);
}
joinNode.setId(ctx_.getNextNodeId());
joinNode.init(analyzer_);
@@ -412,7 +423,14 @@ public class IcebergScanPlanner {
return joinNode;
}
- private void addDataVirtualPositionSlots(TableRef tblRef) throws AnalysisException {
+ /**
+ * Adds the INPUT__FILE__NAME and FILE__POSITION virtual slots to the data scan tuple.
+ * These are needed as join keys by the position-delete join.
+ *
+ * Returns true if the INPUT__FILE__NAME (file path) slot did not exist before this
+ * call, i.e. the planner materialized it solely for the delete join.
+ */
+ private boolean addDataVirtualPositionSlots(TableRef tblRef) throws AnalysisException {
List<String> rawPath = Lists.newArrayList(
tblRef.getUniqueAlias(), VirtualColumn.INPUT_FILE_NAME.getName());
@@ -421,6 +439,19 @@ public class IcebergScanPlanner {
boolean isSemiJoined = analyzer_.isSemiJoined(tblRef.getId());
if (isSemiJoined) analyzer_.setVisibleSemiJoinedTuple(tblRef.getId());
+ // Assume that if the slot was not registered during analyses then this is the only
+ // place where it is used during planning, so it is safe to NULL out the file path
+ // slot after delete processing.
+ Path resolvedPath;
+ try {
+ resolvedPath = analyzer_.resolvePath(rawPath, Path.PathType.SLOT_REF);
+ } catch (TableLoadingException e) {
+ throw new IllegalStateException(e);
+ }
+ Preconditions.checkNotNull(resolvedPath, "resolvedPath is null");
+ boolean filePathCreatedByPlanner = analyzer_.getSlotDescriptor(resolvedPath) == null;
+ // addSlotRefToDesc() reuses the existing slot if already registered, and always marks
+ // it materialized, which the position-delete join requires.
SlotDescriptor fileNameSlotDesc =
SingleNodePlanner.addSlotRefToDesc(analyzer_, rawPath);
fileNameSlotDesc.setStats(virtualInputFileNameStats());
@@ -429,10 +460,10 @@ public class IcebergScanPlanner {
tblRef.getUniqueAlias(), VirtualColumn.FILE_POSITION.getName());
SlotDescriptor filePosSlotDesc =
SingleNodePlanner.addSlotRefToDesc(analyzer_, rawPath);
+ filePosSlotDesc.setStats(virtualFilePositionStats());
if (isSemiJoined) analyzer_.setVisibleSemiJoinedTuple(null);
-
- filePosSlotDesc.setStats(virtualFilePositionStats());
+ return filePathCreatedByPlanner;
}
private void addAllSlotsForEqualityDeletes(TableRef tblRef) throws AnalysisException {
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-optimize.test b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-optimize.test
index 74d3c41fc..39cf32301 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-optimize.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-optimize.test
@@ -66,6 +66,7 @@ WRITE TO HDFS [functional_parquet.iceberg_v2_partitioned_position_deletes, OVERW
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_partitioned_position_deletes.file__position = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.pos, functional_parquet.iceberg_v2_partitioned_position_deletes.input__file__name = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.file_path
+| clear file path slot
| mem-estimate=692B mem-reservation=692B thread-reservation=0
| tuple-ids=0 row-size=64B cardinality=10
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -113,6 +114,7 @@ F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=96.53MB mem-reservation=48.22KB thread-reservation=2
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_partitioned_position_deletes.file__position = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.pos, functional_parquet.iceberg_v2_partitioned_position_deletes.input__file__name = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.file_path
+| clear file path slot
| mem-estimate=230B mem-reservation=230B thread-reservation=0
| tuple-ids=0 row-size=64B cardinality=10
| in pipelines: 00(GETNEXT), 01(OPEN)
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables-resources.test b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables-resources.test
index 26ddb7424..64539dd72 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables-resources.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/iceberg-v2-tables-resources.test
@@ -107,6 +107,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_delete_positional.file__position = functional_parquet.iceberg_v2_delete_positional-position-delete.pos, functional_parquet.iceberg_v2_delete_positional.input__file__name = functional_parquet.iceberg_v2_delete_positional-position-delete.file_path
+| clear file path slot
| mem-estimate=696B mem-reservation=696B thread-reservation=0
| tuple-ids=0 row-size=20B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -160,6 +161,7 @@ Per-Host Resources: mem-estimate=32.02MB mem-reservation=16.68KB thread-reservat
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_delete_positional.file__position = functional_parquet.iceberg_v2_delete_positional-position-delete.pos, functional_parquet.iceberg_v2_delete_positional.input__file__name = functional_parquet.iceberg_v2_delete_positional-position-delete.file_path
+| clear file path slot
| mem-estimate=696B mem-reservation=696B thread-reservation=0
| tuple-ids=0 row-size=20B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -203,6 +205,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_delete_positional.file__position = functional_parquet.iceberg_v2_delete_positional-position-delete.pos, functional_parquet.iceberg_v2_delete_positional.input__file__name = functional_parquet.iceberg_v2_delete_positional-position-delete.file_path
+| clear file path slot
| mem-estimate=696B mem-reservation=696B thread-reservation=0
| tuple-ids=0 row-size=40B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -244,6 +247,7 @@ F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.68KB thread-reservation=2
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_delete_positional.file__position = functional_parquet.iceberg_v2_delete_positional-position-delete.pos, functional_parquet.iceberg_v2_delete_positional.input__file__name = functional_parquet.iceberg_v2_delete_positional-position-delete.file_path
+| clear file path slot
| mem-estimate=696B mem-reservation=696B thread-reservation=0
| tuple-ids=0 row-size=40B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -287,6 +291,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_positional_delete_all_rows.file__position = functional_parquet.iceberg_v2_positional_delete_all_rows-position-delete.pos, functional_parquet.iceberg_v2_positional_delete_all_rows.input__file__name = functional_parquet.iceberg_v2_positional_delete_all_rows-position-delete.file_path
+| clear file path slot
| mem-estimate=764B mem-reservation=764B thread-reservation=0
| tuple-ids=0 row-size=36B cardinality=1
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -328,6 +333,7 @@ F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
Per-Host Resources: mem-estimate=64.16MB mem-reservation=32.75KB thread-reservation=2
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_positional_delete_all_rows.file__position = functional_parquet.iceberg_v2_positional_delete_all_rows-position-delete.pos, functional_parquet.iceberg_v2_positional_delete_all_rows.input__file__name = functional_parquet.iceberg_v2_positional_delete_all_rows-position-delete.file_path
+| clear file path slot
| mem-estimate=764B mem-reservation=764B thread-reservation=0
| tuple-ids=0 row-size=36B cardinality=1
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -371,6 +377,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_positional_delete_all_rows.file__position = functional_parquet.iceberg_v2_positional_delete_all_rows-position-delete.pos, functional_parquet.iceberg_v2_positional_delete_all_rows.input__file__name = functional_parquet.iceberg_v2_positional_delete_all_rows-position-delete.file_path
+| clear file path slot
| limit: 1
| mem-estimate=764B mem-reservation=764B thread-reservation=0
| tuple-ids=0 row-size=36B cardinality=1
@@ -414,6 +421,7 @@ F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
Per-Host Resources: mem-estimate=64.16MB mem-reservation=32.75KB thread-reservation=2
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_positional_delete_all_rows.file__position = functional_parquet.iceberg_v2_positional_delete_all_rows-position-delete.pos, functional_parquet.iceberg_v2_positional_delete_all_rows.input__file__name = functional_parquet.iceberg_v2_positional_delete_all_rows-position-delete.file_path
+| clear file path slot
| limit: 1
| mem-estimate=764B mem-reservation=764B thread-reservation=0
| tuple-ids=0 row-size=36B cardinality=1
@@ -465,6 +473,7 @@ PLAN-ROOT SINK
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=1.53KB mem-reservation=1.53KB thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=2
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -494,6 +503,7 @@ PLAN-ROOT SINK
03:SCAN HDFS [functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files]
HDFS partitions=1/1 files=2 size=1.22KB
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -525,6 +535,7 @@ Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.76KB thread-reservat
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=783B mem-reservation=783B thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=2
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -561,6 +572,7 @@ Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.76KB thread-reservat
03:SCAN HDFS [functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files, RANDOM]
HDFS partitions=1/1 files=2 size=1.22KB
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -585,6 +597,7 @@ PLAN-ROOT SINK
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=1.53KB mem-reservation=1.53KB thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=2
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -614,6 +627,7 @@ PLAN-ROOT SINK
03:SCAN HDFS [functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files]
HDFS partitions=1/1 files=2 size=1.22KB
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -643,6 +657,7 @@ Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.76KB thread-reservat
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=783B mem-reservation=783B thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=2
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -679,6 +694,7 @@ Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.76KB thread-reservat
03:SCAN HDFS [functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files, RANDOM]
HDFS partitions=1/1 files=2 size=1.22KB
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -729,6 +745,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| clear file path slot
| mem-estimate=781B mem-reservation=781B thread-reservation=0
| tuple-ids=0 row-size=20B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -782,6 +799,7 @@ Per-Host Resources: mem-estimate=32.02MB mem-reservation=16.76KB thread-reservat
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| clear file path slot
| mem-estimate=781B mem-reservation=781B thread-reservation=0
| tuple-ids=0 row-size=20B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -831,6 +849,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| clear file path slot
| mem-estimate=781B mem-reservation=781B thread-reservation=0
| tuple-ids=0 row-size=20B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -884,6 +903,7 @@ Per-Host Resources: mem-estimate=32.02MB mem-reservation=16.76KB thread-reservat
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| clear file path slot
| mem-estimate=781B mem-reservation=781B thread-reservation=0
| tuple-ids=0 row-size=20B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -933,6 +953,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| clear file path slot
| mem-estimate=1.53KB mem-reservation=1.53KB thread-reservation=0
| tuple-ids=0 row-size=20B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -986,6 +1007,7 @@ Per-Host Resources: mem-estimate=32.02MB mem-reservation=16.76KB thread-reservat
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| clear file path slot
| mem-estimate=783B mem-reservation=783B thread-reservation=0
| tuple-ids=0 row-size=20B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1035,6 +1057,7 @@ PLAN-ROOT SINK
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_positional_update_all_rows.file__position = functional_parquet.iceberg_v2_positional_update_all_rows-position-delete.pos, functional_parquet.iceberg_v2_positional_update_all_rows.input__file__name = functional_parquet.iceberg_v2_positional_update_all_rows-position-delete.file_path
+| | clear file path slot
| | mem-estimate=764B mem-reservation=764B thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=1
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1064,6 +1087,7 @@ PLAN-ROOT SINK
03:SCAN HDFS [functional_parquet.iceberg_v2_positional_update_all_rows]
HDFS partitions=1/1 files=1 size=625B
Iceberg snapshot id: 3877007445826010687
+ clear file path slot
stored statistics:
table: rows=6 size=3.82KB
columns missing stats: i, s
@@ -1093,6 +1117,7 @@ Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.75KB thread-reservat
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | equality predicates: functional_parquet.iceberg_v2_positional_update_all_rows.file__position = functional_parquet.iceberg_v2_positional_update_all_rows-position-delete.pos, functional_parquet.iceberg_v2_positional_update_all_rows.input__file__name = functional_parquet.iceberg_v2_positional_update_all_rows-position-delete.file_path
+| | clear file path slot
| | mem-estimate=764B mem-reservation=764B thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=1
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1129,6 +1154,7 @@ Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.75KB thread-reservat
03:SCAN HDFS [functional_parquet.iceberg_v2_positional_update_all_rows, RANDOM]
HDFS partitions=1/1 files=1 size=625B
Iceberg snapshot id: 3877007445826010687
+ clear file path slot
stored statistics:
table: rows=6 size=3.82KB
columns missing stats: i, s
@@ -1147,6 +1173,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_partitioned_position_deletes.file__position = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.pos, functional_parquet.iceberg_v2_partitioned_position_deletes.input__file__name = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.file_path
+| clear file path slot
| mem-estimate=2.12KB mem-reservation=2.12KB thread-reservation=0
| tuple-ids=0 row-size=64B cardinality=10
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1188,6 +1215,7 @@ F00:PLAN FRAGMENT [RANDOM] hosts=3 instances=3
Per-Host Resources: mem-estimate=96.27MB mem-reservation=48.71KB thread-reservation=2
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_partitioned_position_deletes.file__position = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.pos, functional_parquet.iceberg_v2_partitioned_position_deletes.input__file__name = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.file_path
+| clear file path slot
| mem-estimate=722B mem-reservation=722B thread-reservation=0
| tuple-ids=0 row-size=64B cardinality=10
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1238,6 +1266,7 @@ PLAN-ROOT SINK
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=1.53KB mem-reservation=1.53KB thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=1
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1271,6 +1300,7 @@ PLAN-ROOT SINK
HDFS partitions=1/1 files=2 size=1.22KB
predicates: i > CAST(2 AS INT)
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -1302,6 +1332,7 @@ Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.76KB thread-reservat
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=783B mem-reservation=783B thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=1
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1342,6 +1373,7 @@ Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.76KB thread-reservat
HDFS partitions=1/1 files=2 size=1.22KB
predicates: i > CAST(2 AS INT)
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -1393,6 +1425,7 @@ PLAN-ROOT SINK
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=1.53KB mem-reservation=1.53KB thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=2
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1422,6 +1455,7 @@ PLAN-ROOT SINK
03:SCAN HDFS [functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files]
HDFS partitions=1/1 files=2 size=1.22KB
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -1494,6 +1528,7 @@ Per-Host Resources: mem-estimate=192.17MB mem-reservation=34.03MB thread-reserva
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=783B mem-reservation=783B thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=2
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1530,6 +1565,7 @@ Per-Host Resources: mem-estimate=192.17MB mem-reservation=34.03MB thread-reserva
03:SCAN HDFS [functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files, RANDOM]
HDFS partitions=1/1 files=2 size=1.22KB
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -1555,6 +1591,7 @@ PLAN-ROOT SINK
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=1.53KB mem-reservation=1.53KB thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=1
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1587,6 +1624,7 @@ PLAN-ROOT SINK
HDFS partitions=1/1 files=2 size=1.22KB
predicates: CAST(i AS BIGINT) + CAST(1000 AS BIGINT) > CAST(1003 AS BIGINT)
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -1617,6 +1655,7 @@ Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.76KB thread-reservat
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=783B mem-reservation=783B thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=1
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1656,6 +1695,7 @@ Per-Host Resources: mem-estimate=64.17MB mem-reservation=32.76KB thread-reservat
HDFS partitions=1/1 files=2 size=1.22KB
predicates: CAST(i AS BIGINT) + CAST(1000 AS BIGINT) > CAST(1003 AS BIGINT)
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -1696,6 +1736,7 @@ PLAN-ROOT SINK
| |
| |--07:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | | equality predicates: functional_parquet.iceberg_v2_positional_update_all_rows.file__position = functional_parquet.iceberg_v2_positional_update_all_rows-position-delete.pos, functional_parquet.iceberg_v2_positional_update_all_rows.input__file__name = functional_parquet.iceberg_v2_positional_update_all_rows-position-delete.file_path
+| | | clear file path slot
| | | mem-estimate=764B mem-reservation=764B thread-reservation=0
| | | tuple-ids=1 row-size=24B cardinality=1
| | | in pipelines: 05(GETNEXT), 06(OPEN)
@@ -1725,6 +1766,7 @@ PLAN-ROOT SINK
| 08:SCAN HDFS [functional_parquet.iceberg_v2_positional_update_all_rows]
| HDFS partitions=1/1 files=1 size=625B
| Iceberg snapshot id: 3877007445826010687
+| clear file path slot
| stored statistics:
| table: rows=6 size=3.82KB
| columns missing stats: i
@@ -1741,6 +1783,7 @@ PLAN-ROOT SINK
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=1.53KB mem-reservation=1.53KB thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=2
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1772,6 +1815,7 @@ PLAN-ROOT SINK
HDFS partitions=1/1 files=2 size=1.22KB
runtime filters: RF000[bloom] -> i
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -1834,6 +1878,7 @@ Per-Host Resources: mem-estimate=67.11MB mem-reservation=2.97MB thread-reservati
| |
| |--07:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | | equality predicates: functional_parquet.iceberg_v2_positional_update_all_rows.file__position = functional_parquet.iceberg_v2_positional_update_all_rows-position-delete.pos, functional_parquet.iceberg_v2_positional_update_all_rows.input__file__name = functional_parquet.iceberg_v2_positional_update_all_rows-position-delete.file_path
+| | | clear file path slot
| | | mem-estimate=764B mem-reservation=764B thread-reservation=0
| | | tuple-ids=1 row-size=24B cardinality=1
| | | in pipelines: 05(GETNEXT), 06(OPEN)
@@ -1870,6 +1915,7 @@ Per-Host Resources: mem-estimate=67.11MB mem-reservation=2.97MB thread-reservati
| 08:SCAN HDFS [functional_parquet.iceberg_v2_positional_update_all_rows, RANDOM]
| HDFS partitions=1/1 files=1 size=625B
| Iceberg snapshot id: 3877007445826010687
+| clear file path slot
| stored statistics:
| table: rows=6 size=3.82KB
| columns missing stats: i
@@ -1886,6 +1932,7 @@ Per-Host Resources: mem-estimate=67.11MB mem-reservation=2.97MB thread-reservati
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=783B mem-reservation=783B thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=2
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1924,6 +1971,7 @@ Per-Host Resources: mem-estimate=67.11MB mem-reservation=2.97MB thread-reservati
HDFS partitions=1/1 files=2 size=1.22KB
runtime filters: RF000[bloom] -> i
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -1942,6 +1990,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_partitioned_position_deletes.file__position = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.pos, functional_parquet.iceberg_v2_partitioned_position_deletes.input__file__name = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.file_path
+| clear file path slot
| mem-estimate=723B mem-reservation=723B thread-reservation=0
| tuple-ids=0 row-size=64B cardinality=1
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -1987,6 +2036,7 @@ F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
Per-Host Resources: mem-estimate=96.27MB mem-reservation=48.71KB thread-reservation=2
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_partitioned_position_deletes.file__position = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.pos, functional_parquet.iceberg_v2_partitioned_position_deletes.input__file__name = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.file_path
+| clear file path slot
| mem-estimate=723B mem-reservation=723B thread-reservation=0
| tuple-ids=0 row-size=64B cardinality=1
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -2035,6 +2085,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_partitioned_position_deletes.file__position = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.pos, functional_parquet.iceberg_v2_partitioned_position_deletes.input__file__name = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.file_path
+| clear file path slot
| mem-estimate=723B mem-reservation=723B thread-reservation=0
| tuple-ids=0 row-size=64B cardinality=1
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -2080,6 +2131,7 @@ F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
Per-Host Resources: mem-estimate=96.27MB mem-reservation=48.71KB thread-reservation=2
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_partitioned_position_deletes.file__position = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.pos, functional_parquet.iceberg_v2_partitioned_position_deletes.input__file__name = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.file_path
+| clear file path slot
| mem-estimate=723B mem-reservation=723B thread-reservation=0
| tuple-ids=0 row-size=64B cardinality=1
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -2128,6 +2180,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_partitioned_position_deletes.file__position = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.pos, functional_parquet.iceberg_v2_partitioned_position_deletes.input__file__name = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.file_path
+| clear file path slot
| mem-estimate=723B mem-reservation=723B thread-reservation=0
| tuple-ids=0 row-size=64B cardinality=4
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -2170,6 +2223,7 @@ F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
Per-Host Resources: mem-estimate=96.27MB mem-reservation=48.71KB thread-reservation=2
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_partitioned_position_deletes.file__position = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.pos, functional_parquet.iceberg_v2_partitioned_position_deletes.input__file__name = functional_parquet.iceberg_v2_partitioned_position_deletes-position-delete.file_path
+| clear file path slot
| mem-estimate=723B mem-reservation=723B thread-reservation=0
| tuple-ids=0 row-size=64B cardinality=4
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -2227,6 +2281,7 @@ PLAN-ROOT SINK
| |
| 07:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=1.53KB mem-reservation=1.53KB thread-reservation=0
| | tuple-ids=4 row-size=20B cardinality=2
| | in pipelines: 05(GETNEXT), 06(OPEN)
@@ -2272,6 +2327,7 @@ PLAN-ROOT SINK
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| clear file path slot
| mem-estimate=1.53KB mem-reservation=1.53KB thread-reservation=0
| tuple-ids=0 row-size=20B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -2337,6 +2393,7 @@ PLAN-ROOT SINK
| |
| 07:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=783B mem-reservation=783B thread-reservation=0
| | tuple-ids=4 row-size=20B cardinality=2
| | in pipelines: 05(GETNEXT), 06(OPEN)
@@ -2409,6 +2466,7 @@ Per-Host Resources: mem-estimate=32.02MB mem-reservation=16.76KB thread-reservat
|
02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| clear file path slot
| mem-estimate=783B mem-reservation=783B thread-reservation=0
| tuple-ids=0 row-size=20B cardinality=2
| in pipelines: 00(GETNEXT), 01(OPEN)
@@ -2717,6 +2775,7 @@ PLAN-ROOT SINK
| |
| 02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_delete_both_eq_and_pos.file__position = functional_parquet.iceberg_v2_delete_both_eq_and_pos-position-delete.pos, functional_parquet.iceberg_v2_delete_both_eq_and_pos.input__file__name = functional_parquet.iceberg_v2_delete_both_eq_and_pos-position-delete.file_path
+| | clear file path slot
| | mem-estimate=701B mem-reservation=701B thread-reservation=0
| | tuple-ids=0 row-size=48B cardinality=1
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -2746,6 +2805,7 @@ PLAN-ROOT SINK
05:SCAN HDFS [functional_parquet.iceberg_v2_delete_both_eq_and_pos]
HDFS partitions=1/1 files=1 size=885B
Iceberg snapshot id: 911559291487642581
+ clear file path slot
stored statistics:
table: rows=4 size=3.95KB
columns missing stats: i, s, d
@@ -2800,6 +2860,7 @@ Per-Host Resources: mem-estimate=98.17MB mem-reservation=1.99MB thread-reservati
| |
| 02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | equality predicates: functional_parquet.iceberg_v2_delete_both_eq_and_pos.file__position = functional_parquet.iceberg_v2_delete_both_eq_and_pos-position-delete.pos, functional_parquet.iceberg_v2_delete_both_eq_and_pos.input__file__name = functional_parquet.iceberg_v2_delete_both_eq_and_pos-position-delete.file_path
+| | clear file path slot
| | mem-estimate=701B mem-reservation=701B thread-reservation=0
| | tuple-ids=0 row-size=48B cardinality=1
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -2836,6 +2897,7 @@ Per-Host Resources: mem-estimate=98.17MB mem-reservation=1.99MB thread-reservati
05:SCAN HDFS [functional_parquet.iceberg_v2_delete_both_eq_and_pos, RANDOM]
HDFS partitions=1/1 files=1 size=885B
Iceberg snapshot id: 911559291487642581
+ clear file path slot
stored statistics:
table: rows=4 size=3.95KB
columns missing stats: i, s, d
@@ -3401,6 +3463,7 @@ PLAN-ROOT SINK
| |
| 02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_delete_pos_and_multi_eq_ids.file__position = functional_parquet.iceberg_v2_delete_pos_and_multi_eq_ids-position-delete.pos, functional_parquet.iceberg_v2_delete_pos_and_multi_eq_ids.input__file__name = functional_parquet.iceberg_v2_delete_pos_and_multi_eq_ids-position-delete.file_path
+| | clear file path slot
| | mem-estimate=1.38KB mem-reservation=1.38KB thread-reservation=0
| | tuple-ids=0 row-size=48B cardinality=4
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -3430,6 +3493,7 @@ PLAN-ROOT SINK
07:SCAN HDFS [functional_parquet.iceberg_v2_delete_pos_and_multi_eq_ids]
HDFS partitions=1/1 files=1 size=853B
Iceberg snapshot id: 152862018760071153
+ clear file path slot
stored statistics:
table: rows=7 size=5.39KB
columns missing stats: i, s, d
@@ -3509,6 +3573,7 @@ Per-Host Resources: mem-estimate=100.13MB mem-reservation=3.92MB thread-reservat
| |
| 02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN, DIRECTED]
| | equality predicates: functional_parquet.iceberg_v2_delete_pos_and_multi_eq_ids.file__position = functional_parquet.iceberg_v2_delete_pos_and_multi_eq_ids-position-delete.pos, functional_parquet.iceberg_v2_delete_pos_and_multi_eq_ids.input__file__name = functional_parquet.iceberg_v2_delete_pos_and_multi_eq_ids-position-delete.file_path
+| | clear file path slot
| | mem-estimate=705B mem-reservation=705B thread-reservation=0
| | tuple-ids=0 row-size=48B cardinality=4
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -3545,6 +3610,7 @@ Per-Host Resources: mem-estimate=100.13MB mem-reservation=3.92MB thread-reservat
07:SCAN HDFS [functional_parquet.iceberg_v2_delete_pos_and_multi_eq_ids, RANDOM]
HDFS partitions=1/1 files=1 size=853B
Iceberg snapshot id: 152862018760071153
+ clear file path slot
stored statistics:
table: rows=7 size=5.39KB
columns missing stats: i, s, d
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tablesample-iceberg.test b/testdata/workloads/functional-planner/queries/PlannerTest/tablesample-iceberg.test
index 2c0e3341c..1b99134fb 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tablesample-iceberg.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tablesample-iceberg.test
@@ -109,6 +109,7 @@ PLAN-ROOT SINK
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.file__position = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.pos, functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files.input__file__name = functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files-position-delete.file_path
+| | clear file path slot
| | mem-estimate=566B mem-reservation=566B thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=3
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -138,6 +139,7 @@ PLAN-ROOT SINK
03:SCAN HDFS [functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files]
HDFS partitions=1/1 files=1 size=620B
Iceberg snapshot id: 1497619269847778439
+ clear file path slot
stored statistics:
table: rows=10 size=7.77KB
columns missing stats: i, s
@@ -167,6 +169,7 @@ PLAN-ROOT SINK
|
|--02:DELETE EVENTS ICEBERG DELETE [ICEBERG DELETE JOIN]
| | equality predicates: functional_parquet.iceberg_v2_positional_update_all_rows.file__position = functional_parquet.iceberg_v2_positional_update_all_rows-position-delete.pos, functional_parquet.iceberg_v2_positional_update_all_rows.input__file__name = functional_parquet.iceberg_v2_positional_update_all_rows-position-delete.file_path
+| | clear file path slot
| | mem-estimate=764B mem-reservation=764B thread-reservation=0
| | tuple-ids=0 row-size=36B cardinality=2
| | in pipelines: 00(GETNEXT), 01(OPEN)
@@ -196,6 +199,7 @@ PLAN-ROOT SINK
03:SCAN HDFS [functional_parquet.iceberg_v2_positional_update_all_rows]
HDFS partitions=1/1 files=1 size=625B
Iceberg snapshot id: 3877007445826010687
+ clear file path slot
stored statistics:
table: rows=6 size=3.82KB
columns missing stats: i, s
diff --git a/tests/query_test/test_iceberg.py b/tests/query_test/test_iceberg.py
index 1fc16a3de..0284e3b74 100644
--- a/tests/query_test/test_iceberg.py
+++ b/tests/query_test/test_iceberg.py
@@ -43,9 +43,10 @@ from tests.common.file_utils import (
)
from tests.common.iceberg_test_suite import IcebergTestSuite
from tests.common.impala_connection import IMPALA_CONNECTION_EXCEPTION
-from tests.common.skip import SkipIf, SkipIfDockerizedCluster, SkipIfFS
+from tests.common.skip import (SkipIf, SkipIfDockerizedCluster, SkipIfFS,
+ SkipIfNotHdfsMinicluster)
from tests.common.test_dimensions import (add_exec_option_dimension,
- create_exec_option_dimension)
+ create_exec_option_dimension, create_single_exec_option_dimension)
from tests.common.test_result_verifier import error_msg_startswith
from tests.shell.util import run_impala_shell_cmd
from tests.util.filesystem_utils import FILESYSTEM_PREFIX, get_fs_path, IS_HDFS, WAREHOUSE
@@ -2948,3 +2949,57 @@ class TestIcebergTableWithPuffinStats(IcebergTestSuite):
def _get_ndvs_from_query_result(self, query_result):
rows = query_result.get_data().split("\n")
return [int(row.split()[2]) for row in rows]
+
+
+class TestIcebergDeleteFilePathSlot(IcebergTestSuite):
+ """IMPALA-15171: the delete-join file path slot is nulled when nothing above the join
+ needs it, so the (long) file path string is not propagated through exchanges. A
+ self-join over a table with deletes should send < 250 MB and keep the hash join under
+ 250 MB; without the fix these balloon to ~1 GB / ~590 MB."""
+
+ @classmethod
+ def add_test_dimensions(cls):
+ super(TestIcebergDeleteFilePathSlot, cls).add_test_dimensions()
+ # Heavy CTAS + self-join: run once, with a fixed batch_size the thresholds depend on.
+ cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension())
+ cls.ImpalaTestMatrix.add_constraint(
+ lambda v: v.get_value('table_format').file_format == 'parquet')
+
+ @SkipIfNotHdfsMinicluster.tuned_for_minicluster
+ def test_delete_file_path_slot_not_propagated(self, unique_database):
+ tbl = unique_database + ".ice_lineitem"
+ self.execute_query("create table {0} stored as iceberg as select * from "
+ "tpch_parquet.lineitem".format(tbl))
+ # Single-row delete: we will have data files with and without deletes as well, i.e.
+ # below query exercises both the with-deletes and the without-deletes branches
+ self.execute_query("delete from {0} where l_orderkey=1000000".format(tbl))
+
+ query = ("select count(*) from {0} a, {0} b "
+ "where a.l_orderkey=b.l_orderkey".format(tbl))
+ result = self.client.execute(query, fetch_exec_summary=True)
+
+ # Correctness (also guards against nulling corrupting the shared probe tuples).
+ EXPECTED_SELF_JOIN_COUNT = 30012984
+ assert result.data == [str(EXPECTED_SELF_JOIN_COUNT)], result.data
+
+ # Bound for TotalBytesSent and hash join peak mem, tuned for the 3-node minicluster.
+ MAX_BYTES = 250 * 1024 * 1024
+
+ # First TotalBytesSent in the profile is the query-level counter.
+ match = re.search(r"- TotalBytesSent: [\d.]+ [KMGT]?B \((\d+)\)",
+ result.runtime_profile)
+ assert match is not None, result.runtime_profile
+ total_bytes_sent = int(match.group(1))
+ assert total_bytes_sent < MAX_BYTES, (
+ "TotalBytesSent={0} bytes, expected < {1}.\n{2}".format(
+ total_bytes_sent, MAX_BYTES, result.runtime_profile))
+
+ # exec_summary 'peak_mem' is bytes; only the self-join matches "HASH JOIN" (the delete
+ # joins are labelled "... ICEBERG DELETE").
+ hash_joins = [row for row in result.exec_summary
+ if row['operator'].endswith('HASH JOIN')]
+ assert len(hash_joins) == 1, result.exec_summary
+ hash_join_peak_mem = hash_joins[0]['peak_mem']
+ assert hash_join_peak_mem < MAX_BYTES, (
+ "HASH JOIN peak_mem={0} bytes, expected < {1}.\n{2}".format(
+ hash_join_peak_mem, MAX_BYTES, result.exec_summary))
(impala) 02/02: IMPALA-14684: Add metrics for catalogd table invalidations
Posted by mi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 538ceb4223d1ceffb525c6fa355de3d459d6c585
Author: Arnab Karmakar <ar...@gmail.com>
AuthorDate: Wed Apr 1 04:19:41 2026 -0700
IMPALA-14684: Add metrics for catalogd table invalidations
Implements sliding window metrics to track table invalidations due to
TTL and memory pressure, helping administrators monitor and tune
configurations like invalidate_tables_timeout_s and JVM heap size.
New metrics added:
- catalog.num-ttl-invalidated-tables: Total TTL-based invalidations
- catalog.num-memory-pressure-invalidated-tables: Total memory pressure
invalidations
- catalog.ttl-invalidations-{10s,01m,05m,30m}: TTL invalidations in
sliding windows
- catalog.memory-pressure-invalidations-{10s,01m,05m,30m}: Memory
pressure invalidations in sliding windows
- catalog.last-ttl-invalidated-tables: Tables invalidated by last TTL
invalidation batch
- catalog.last-ttl-invalidation-ms: Milliseconds since epoch of last TTL
invalidation batch
- catalog.last-memory-pressure-invalidated-tables: Tables invalidated by
last memory pressure invalidation batch
- catalog.last-memory-pressure-invalidation-ms: Milliseconds since epoch
of last memory pressure invalidation batch
Changes:
- CatalogdTableInvalidator tracks timestamps, Catalog provides counters
- Added 14 new fields to TGetCatalogServerMetricsResponse
- Updated CatalogServer to register and refresh new metrics
Testing:
- Added JUnit tests that verify timestamp queue operations and binary
search efficiency
- Added integration test that validates TTL-based invalidation and sliding
window metrics
Change-Id: If78e6c8429801e3c1829860666f04ad287b826ab
Assisted-by: composer-2.5-fast
Reviewed-on: http://gerrit.cloudera.org:8080/24166
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/catalog/catalog-server.cc | 85 +++++++++
be/src/catalog/catalog-server.h | 25 +++
common/thrift/JniCatalog.thrift | 24 +++
common/thrift/metrics.json | 140 ++++++++++++++
.../java/org/apache/impala/catalog/Catalog.java | 40 +++-
.../impala/catalog/CatalogServiceCatalog.java | 2 +-
.../impala/catalog/CatalogdTableInvalidator.java | 172 ++++++++++++++++++
.../java/org/apache/impala/service/JniCatalog.java | 32 ++++
.../catalog/CatalogdTableInvalidatorTest.java | 87 +++++++++
.../custom_cluster/test_automatic_invalidation.py | 201 +++++++++++++++++++++
10 files changed, 806 insertions(+), 2 deletions(-)
diff --git a/be/src/catalog/catalog-server.cc b/be/src/catalog/catalog-server.cc
index 96d5d922f..ace143424 100644
--- a/be/src/catalog/catalog-server.cc
+++ b/be/src/catalog/catalog-server.cc
@@ -379,6 +379,36 @@ const string CATALOG_NUM_LOADED_TABLES = "catalog.num-loaded-tables";
const string CATALOG_NUM_HMS_CLIENTS_IDLE = "catalog.hms-client-pool.num-idle";
const string CATALOG_NUM_HMS_CLIENTS_IN_USE = "catalog.hms-client-pool.num-in-use";
+// Table invalidation metrics
+const string CATALOG_NUM_TTL_INVALIDATED_TABLES =
+ "catalog.num-ttl-invalidated-tables";
+const string CATALOG_NUM_MEMORY_PRESSURE_INVALIDATED_TABLES =
+ "catalog.num-memory-pressure-invalidated-tables";
+const string CATALOG_TTL_INVALIDATIONS_10S =
+ "catalog.ttl-invalidations-10s";
+const string CATALOG_TTL_INVALIDATIONS_1M =
+ "catalog.ttl-invalidations-01m";
+const string CATALOG_TTL_INVALIDATIONS_5M =
+ "catalog.ttl-invalidations-05m";
+const string CATALOG_TTL_INVALIDATIONS_30M =
+ "catalog.ttl-invalidations-30m";
+const string CATALOG_LAST_TTL_INVALIDATION_MS =
+ "catalog.last-ttl-invalidation-ms";
+const string CATALOG_LAST_TTL_INVALIDATED_TABLES =
+ "catalog.last-ttl-invalidated-tables";
+const string CATALOG_MEMORY_PRESSURE_INVALIDATIONS_10S =
+ "catalog.memory-pressure-invalidations-10s";
+const string CATALOG_MEMORY_PRESSURE_INVALIDATIONS_1M =
+ "catalog.memory-pressure-invalidations-01m";
+const string CATALOG_MEMORY_PRESSURE_INVALIDATIONS_5M =
+ "catalog.memory-pressure-invalidations-05m";
+const string CATALOG_MEMORY_PRESSURE_INVALIDATIONS_30M =
+ "catalog.memory-pressure-invalidations-30m";
+const string CATALOG_LAST_MEMORY_PRESSURE_INVALIDATION_MS =
+ "catalog.last-memory-pressure-invalidation-ms";
+const string CATALOG_LAST_MEMORY_PRESSURE_INVALIDATED_TABLES =
+ "catalog.last-memory-pressure-invalidated-tables";
+
const string CATALOG_WEB_PAGE = "/catalog";
const string CATALOG_TEMPLATE = "catalog.tmpl";
const string CATALOG_OBJECT_WEB_PAGE = "/catalog_object";
@@ -661,6 +691,36 @@ CatalogServer::CatalogServer(MetricGroup* metrics)
num_hms_clients_idle_metric_ = metrics->AddGauge(CATALOG_NUM_HMS_CLIENTS_IDLE, 0);
num_hms_clients_in_use_metric_ = metrics->AddGauge(CATALOG_NUM_HMS_CLIENTS_IN_USE, 0);
+ num_ttl_invalidated_tables_metric_ =
+ metrics->AddGauge(CATALOG_NUM_TTL_INVALIDATED_TABLES, 0);
+ num_memory_pressure_invalidated_tables_metric_ =
+ metrics->AddGauge(CATALOG_NUM_MEMORY_PRESSURE_INVALIDATED_TABLES, 0);
+
+ ttl_invalidations_10s_metric_ =
+ metrics->AddGauge(CATALOG_TTL_INVALIDATIONS_10S, 0);
+ ttl_invalidations_1m_metric_ =
+ metrics->AddGauge(CATALOG_TTL_INVALIDATIONS_1M, 0);
+ ttl_invalidations_5m_metric_ =
+ metrics->AddGauge(CATALOG_TTL_INVALIDATIONS_5M, 0);
+ ttl_invalidations_30m_metric_ =
+ metrics->AddGauge(CATALOG_TTL_INVALIDATIONS_30M, 0);
+ last_ttl_invalidation_ms_metric_ =
+ metrics->AddGauge(CATALOG_LAST_TTL_INVALIDATION_MS, 0);
+ last_ttl_invalidated_tables_metric_ =
+ metrics->AddGauge(CATALOG_LAST_TTL_INVALIDATED_TABLES, 0);
+ memory_pressure_invalidations_10s_metric_ =
+ metrics->AddGauge(CATALOG_MEMORY_PRESSURE_INVALIDATIONS_10S, 0);
+ memory_pressure_invalidations_1m_metric_ =
+ metrics->AddGauge(CATALOG_MEMORY_PRESSURE_INVALIDATIONS_1M, 0);
+ memory_pressure_invalidations_5m_metric_ =
+ metrics->AddGauge(CATALOG_MEMORY_PRESSURE_INVALIDATIONS_5M, 0);
+ memory_pressure_invalidations_30m_metric_ =
+ metrics->AddGauge(CATALOG_MEMORY_PRESSURE_INVALIDATIONS_30M, 0);
+ last_memory_pressure_invalidation_ms_metric_ =
+ metrics->AddGauge(CATALOG_LAST_MEMORY_PRESSURE_INVALIDATION_MS, 0);
+ last_memory_pressure_invalidated_tables_metric_ =
+ metrics->AddGauge(CATALOG_LAST_MEMORY_PRESSURE_INVALIDATED_TABLES, 0);
+
active_status_metric_ =
metrics->AddProperty(CATALOG_ACTIVE_STATUS, !FLAGS_enable_catalogd_ha);
num_ha_active_status_change_metric_ =
@@ -1142,6 +1202,31 @@ void CatalogServer::MarkPendingMetadataReset(const unique_lock<std::mutex>& lock
num_loaded_tables_metric_->SetValue(response.catalog_num_loaded_tables);
num_hms_clients_idle_metric_->SetValue(response.catalog_num_hms_clients_idle);
num_hms_clients_in_use_metric_->SetValue(response.catalog_num_hms_clients_in_use);
+
+ num_ttl_invalidated_tables_metric_->SetValue(
+ response.catalog_num_ttl_invalidated_tables);
+ num_memory_pressure_invalidated_tables_metric_->SetValue(
+ response.catalog_num_memory_pressure_invalidated_tables);
+ ttl_invalidations_10s_metric_->SetValue(response.catalog_ttl_invalidations_10s);
+ ttl_invalidations_1m_metric_->SetValue(response.catalog_ttl_invalidations_1m);
+ ttl_invalidations_5m_metric_->SetValue(response.catalog_ttl_invalidations_5m);
+ ttl_invalidations_30m_metric_->SetValue(response.catalog_ttl_invalidations_30m);
+ last_ttl_invalidation_ms_metric_->SetValue(response.catalog_last_ttl_invalidation_ms);
+ last_ttl_invalidated_tables_metric_->SetValue(
+ response.catalog_last_ttl_invalidated_tables);
+ memory_pressure_invalidations_10s_metric_->SetValue(
+ response.catalog_memory_pressure_invalidations_10s);
+ memory_pressure_invalidations_1m_metric_->SetValue(
+ response.catalog_memory_pressure_invalidations_1m);
+ memory_pressure_invalidations_5m_metric_->SetValue(
+ response.catalog_memory_pressure_invalidations_5m);
+ memory_pressure_invalidations_30m_metric_->SetValue(
+ response.catalog_memory_pressure_invalidations_30m);
+ last_memory_pressure_invalidation_ms_metric_->SetValue(
+ response.catalog_last_memory_pressure_invalidation_ms);
+ last_memory_pressure_invalidated_tables_metric_->SetValue(
+ response.catalog_last_memory_pressure_invalidated_tables);
+
TEventProcessorMetrics eventProcessorMetrics = response.event_metrics;
MetastoreEventMetrics::refresh(&eventProcessorMetrics);
}
diff --git a/be/src/catalog/catalog-server.h b/be/src/catalog/catalog-server.h
index 248c8f0d6..4bd1c44de 100644
--- a/be/src/catalog/catalog-server.h
+++ b/be/src/catalog/catalog-server.h
@@ -213,6 +213,31 @@ class CatalogServer {
IntGauge* num_hms_clients_idle_metric_;
IntGauge* num_hms_clients_in_use_metric_;
+ /// Metrics that track table invalidations
+ /// Total number of tables invalidated due to TTL
+ IntGauge* num_ttl_invalidated_tables_metric_;
+ /// Total number of tables invalidated due to memory pressure
+ IntGauge* num_memory_pressure_invalidated_tables_metric_;
+
+ /// Sliding window counts for invalidations (10s/1m/5m/30m)
+ IntGauge* ttl_invalidations_10s_metric_;
+ IntGauge* ttl_invalidations_1m_metric_;
+ IntGauge* ttl_invalidations_5m_metric_;
+ IntGauge* ttl_invalidations_30m_metric_;
+ IntGauge* memory_pressure_invalidations_10s_metric_;
+ IntGauge* memory_pressure_invalidations_1m_metric_;
+ IntGauge* memory_pressure_invalidations_5m_metric_;
+ IntGauge* memory_pressure_invalidations_30m_metric_;
+
+ /// Metrics that track the last TTL invalidation and the number of tables invalidated.
+ IntGauge* last_ttl_invalidation_ms_metric_;
+ IntGauge* last_ttl_invalidated_tables_metric_;
+
+ /// Metrics that track the last memory pressure invalidation and
+ /// the number of tables invalidated.
+ IntGauge* last_memory_pressure_invalidation_ms_metric_;
+ IntGauge* last_memory_pressure_invalidated_tables_metric_;
+
/// Thread that polls and execute first reset metadata operation.
std::unique_ptr<Thread> catalog_first_reset_metadata_thread_;
diff --git a/common/thrift/JniCatalog.thrift b/common/thrift/JniCatalog.thrift
index 704e2e315..a8f3e68eb 100644
--- a/common/thrift/JniCatalog.thrift
+++ b/common/thrift/JniCatalog.thrift
@@ -1125,6 +1125,30 @@ struct TGetCatalogServerMetricsResponse {
// Number of loaded tables (i.e. not IncompleteTable)
15: optional i32 catalog_num_loaded_tables
+
+ // Table invalidation metrics
+ // Total number of tables invalidated due to TTL
+ 16: optional i64 catalog_num_ttl_invalidated_tables
+ // Total number of tables invalidated due to memory pressure
+ 17: optional i64 catalog_num_memory_pressure_invalidated_tables
+
+ // Sliding window counts for TTL-based invalidations (10s/1m/5m/30m)
+ 18: optional i64 catalog_ttl_invalidations_10s
+ 19: optional i64 catalog_ttl_invalidations_1m
+ 20: optional i64 catalog_ttl_invalidations_5m
+ 21: optional i64 catalog_ttl_invalidations_30m
+
+ // Sliding window counts for memory pressure-based invalidations (10s/1m/5m/30m)
+ 22: optional i64 catalog_memory_pressure_invalidations_10s
+ 23: optional i64 catalog_memory_pressure_invalidations_1m
+ 24: optional i64 catalog_memory_pressure_invalidations_5m
+ 25: optional i64 catalog_memory_pressure_invalidations_30m
+
+ // Last invalidation batch details
+ 26: optional i64 catalog_last_ttl_invalidation_ms
+ 27: optional i64 catalog_last_ttl_invalidated_tables
+ 28: optional i64 catalog_last_memory_pressure_invalidation_ms
+ 29: optional i64 catalog_last_memory_pressure_invalidated_tables
}
// Request to copy the generated testcase from a given input path.
diff --git a/common/thrift/metrics.json b/common/thrift/metrics.json
index 7763def56..ee8cd3a15 100644
--- a/common/thrift/metrics.json
+++ b/common/thrift/metrics.json
@@ -601,6 +601,146 @@
"kind": "GAUGE",
"key": "catalog.hms-client-pool.num-in-use"
},
+ {
+ "description": "Total number of tables invalidated due to TTL (Time-To-Live).",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "TTL Invalidated Tables",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.num-ttl-invalidated-tables"
+ },
+ {
+ "description": "Total number of tables invalidated due to memory pressure.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Memory Pressure Invalidated Tables",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.num-memory-pressure-invalidated-tables"
+ },
+ {
+ "description": "Number of tables invalidated due to TTL in the last 10 seconds.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "TTL Invalidations (10s)",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.ttl-invalidations-10s"
+ },
+ {
+ "description": "Number of tables invalidated due to TTL in the last 1 minute.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "TTL Invalidations (1m)",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.ttl-invalidations-01m"
+ },
+ {
+ "description": "Number of tables invalidated due to TTL in the last 5 minutes.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "TTL Invalidations (5m)",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.ttl-invalidations-05m"
+ },
+ {
+ "description": "Number of tables invalidated due to TTL in the last 30 minutes.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "TTL Invalidations (30m)",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.ttl-invalidations-30m"
+ },
+ {
+ "description": "Unix timestamp in milliseconds when the last TTL invalidation batch ran.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Last TTL invalidation (ms)",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.last-ttl-invalidation-ms"
+ },
+ {
+ "description": "Number of tables invalidated in the last TTL invalidation batch.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Last TTL invalidated tables",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.last-ttl-invalidated-tables"
+ },
+ {
+ "description": "Number of tables invalidated due to memory pressure in the last 10 seconds.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Memory Pressure Invalidations (10s)",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.memory-pressure-invalidations-10s"
+ },
+ {
+ "description": "Number of tables invalidated due to memory pressure in the last 1 minute.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Memory Pressure Invalidations (1m)",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.memory-pressure-invalidations-01m"
+ },
+ {
+ "description": "Number of tables invalidated due to memory pressure in the last 5 minutes.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Memory Pressure Invalidations (5m)",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.memory-pressure-invalidations-05m"
+ },
+ {
+ "description": "Number of tables invalidated due to memory pressure in the last 30 minutes.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Memory Pressure Invalidations (30m)",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.memory-pressure-invalidations-30m"
+ },
+ {
+ "description": "Unix timestamp in milliseconds when the last memory-pressure invalidation batch ran.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Last memory-pressure invalidation (ms)",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.last-memory-pressure-invalidation-ms"
+ },
+ {
+ "description": "Number of tables invalidated in the last memory-pressure invalidation batch.",
+ "contexts": [
+ "CATALOGSERVER"
+ ],
+ "label": "Last memory-pressure invalidated tables",
+ "units": "NONE",
+ "kind": "GAUGE",
+ "key": "catalog.last-memory-pressure-invalidated-tables"
+ },
{
"description": "Catalog topic update version.",
"contexts": [
diff --git a/fe/src/main/java/org/apache/impala/catalog/Catalog.java b/fe/src/main/java/org/apache/impala/catalog/Catalog.java
index 096b570ac..f1b4b7663 100644
--- a/fe/src/main/java/org/apache/impala/catalog/Catalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/Catalog.java
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
@@ -131,10 +132,17 @@ public abstract class Catalog implements AutoCloseable {
// Number of loaded tables (i.e. not IncompleteTable) in the catalog cache.
// This is updated in real-time when tables are loaded or invalidated.
- // IMPALA-13863: Track loaded tables to help monitor catalog memory pressure.
+ // Track loaded tables to help monitor catalog memory pressure.
// This is static because it tracks loaded tables across all catalog instances.
protected static final AtomicInteger numLoadedTables_ = new AtomicInteger(0);
+ // Number of tables invalidated due to TTL (Time-To-Live).
+ protected static final AtomicLong numTtlInvalidatedTables_ = new AtomicLong(0);
+
+ // Number of tables invalidated due to memory pressure.
+ protected static final AtomicLong numMemoryPressureInvalidatedTables_ =
+ new AtomicLong(0);
+
// Update function for tracking table additions/replacements in the loaded tables
// counter. This is passed to Db#addTable() to centralize the counter update logic.
public static final BiConsumer<Table, Table> TABLE_ADD_UPDATE_FUNC =
@@ -365,6 +373,36 @@ public abstract class Catalog implements AutoCloseable {
return numLoadedTables_.get();
}
+ /**
+ * Returns the total number of tables invalidated due to TTL.
+ */
+ public long getNumTtlInvalidatedTables() {
+ return numTtlInvalidatedTables_.get();
+ }
+
+ /**
+ * Returns the total number of tables invalidated due to memory pressure.
+ */
+ public long getNumMemoryPressureInvalidatedTables() {
+ return numMemoryPressureInvalidatedTables_.get();
+ }
+
+ /**
+ * Increments the counter for TTL-based invalidations by {@code n}.
+ */
+ public static void incrementTtlInvalidatedTables(int n) {
+ Preconditions.checkArgument(n >= 0, "n must be non-negative");
+ numTtlInvalidatedTables_.addAndGet(n);
+ }
+
+ /**
+ * Increments the counter for memory pressure-based invalidations by {@code n}.
+ */
+ public static void incrementMemoryPressureInvalidatedTables(int n) {
+ Preconditions.checkArgument(n >= 0, "n must be non-negative");
+ numMemoryPressureInvalidatedTables_.addAndGet(n);
+ }
+
/**
* Returns a list of data sources names that match pattern.
*
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index 9519fb025..89d887b47 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -4919,7 +4919,7 @@ public class CatalogServiceCatalog extends Catalog {
}
}
- CatalogdTableInvalidator getCatalogdTableInvalidator() {
+ public CatalogdTableInvalidator getCatalogdTableInvalidator() {
return catalogdTableInvalidator_;
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogdTableInvalidator.java b/fe/src/main/java/org/apache/impala/catalog/CatalogdTableInvalidator.java
index 5d214f36a..b15b4e14f 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogdTableInvalidator.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogdTableInvalidator.java
@@ -34,9 +34,11 @@ import javax.management.Notification;
import javax.management.NotificationEmitter;
import javax.management.NotificationListener;
import java.lang.management.MemoryUsage;
+import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
+import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
@@ -53,6 +55,10 @@ import java.util.concurrent.atomic.AtomicLong;
public class CatalogdTableInvalidator {
public static final Logger LOG =
LoggerFactory.getLogger(CatalogdTableInvalidator.class);
+
+ // Maximum retention time for invalidation batch history (30 minutes)
+ private static final long MAX_TIMESTAMP_RETENTION_MS = TimeUnit.MINUTES.toMillis(30);
+
/**
* Plugable time source for tests. Defined as static to avoid passing
* CatalogdTableInvalidator everywhere the clock is used.
@@ -97,6 +103,18 @@ public class CatalogdTableInvalidator {
*/
private long lastInvalidationTime_;
+ /**
+ * One entry per invalidation batch: wall-clock timestamp in millis and table count.
+ */
+ private record InvalidationBatch(long timestampMillis, int count) {}
+
+ private final ArrayDeque<InvalidationBatch> ttlInvalidationBatches_ =
+ new ArrayDeque<>();
+ private final ArrayDeque<InvalidationBatch> memoryPressureInvalidationBatches_ =
+ new ArrayDeque<>();
+ private InvalidationBatch lastTtlInvalidationBatch_;
+ private InvalidationBatch lastMemoryPressureInvalidationBatch_;
+
CatalogdTableInvalidator(CatalogServiceCatalog catalog, final long unusedTableTtlSec,
boolean invalidateTableOnMemoryPressure, double oldGenFullThreshold,
double gcInvalidationFraction) {
@@ -214,6 +232,7 @@ public class CatalogdTableInvalidator {
return Long.compare(o1.getLastUsedTime(), o2.getLastUsedTime());
}
});
+ int numInvalidated = 0;
for (int i = 0; i < tables.size() * invalidationFraction; ++i) {
TTableName tTableName = tables.get(i).getTableName().toThrift();
Reference<Boolean> tblWasRemoved = new Reference<>();
@@ -222,11 +241,17 @@ public class CatalogdTableInvalidator {
NoOpEventSequence.INSTANCE);
LOG.info("Table " + tables.get(i).getFullName() + " invalidated due to memory " +
"pressure.");
+ numInvalidated++;
+ }
+ // Track memory pressure invalidations
+ if (numInvalidated > 0) {
+ recordMemoryPressureInvalidationBatch(numInvalidated);
}
}
private void invalidateOlderThan(long retireAgeNano) {
long now = TIME_SOURCE.read();
+ int numInvalidated = 0;
for (Db db : catalog_.getAllDbs()) {
for (Table table : catalog_.getAllTables(db)) {
if (table instanceof IncompleteTable) continue;
@@ -240,8 +265,13 @@ public class CatalogdTableInvalidator {
LOG.info(
"Invalidated " + table.getFullName() + " due to inactivity for " +
TimeUnit.NANOSECONDS.toSeconds(inactivityTime) + " seconds.");
+ numInvalidated++;
}
}
+ // Track TTL invalidations
+ if (numInvalidated > 0) {
+ recordTtlInvalidationBatch(numInvalidated);
+ }
}
void stop() {
@@ -261,6 +291,148 @@ public class CatalogdTableInvalidator {
notify();
}
+ private InvalidationBatch appendBatch(
+ ArrayDeque<InvalidationBatch> batches, int count) {
+ long now = System.currentTimeMillis();
+ InvalidationBatch batch = new InvalidationBatch(now, count);
+ batches.addLast(batch);
+ long cutoffTime = now - MAX_TIMESTAMP_RETENTION_MS;
+ while (!batches.isEmpty() && batches.peekFirst().timestampMillis() < cutoffTime) {
+ batches.removeFirst();
+ }
+ return batch;
+ }
+
+ private void recordTtlInvalidationBatch(int count) {
+ synchronized (ttlInvalidationBatches_) {
+ lastTtlInvalidationBatch_ = appendBatch(ttlInvalidationBatches_, count);
+ }
+ Catalog.incrementTtlInvalidatedTables(count);
+ }
+
+ private void recordMemoryPressureInvalidationBatch(int count) {
+ synchronized (memoryPressureInvalidationBatches_) {
+ lastMemoryPressureInvalidationBatch_ =
+ appendBatch(memoryPressureInvalidationBatches_, count);
+ }
+ Catalog.incrementMemoryPressureInvalidatedTables(count);
+ }
+
+ /**
+ * Sliding-window invalidation counts for 10s / 1m / 5m / 30m.
+ */
+ private record SlidingWindowCounts(
+ long invalidations10Sec,
+ long invalidations1Min,
+ long invalidations5Min,
+ long invalidations30Min) {}
+
+ /**
+ * Counts invalidations in all sliding windows with a single pass.
+ * Caller must hold the lock on {@code batches}.
+ */
+ private SlidingWindowCounts countInvalidationsInSlidingWindows(
+ ArrayDeque<InvalidationBatch> batches) {
+ long now = System.currentTimeMillis();
+ long cutoff10Sec = now - TimeUnit.SECONDS.toMillis(10);
+ long cutoff1Min = now - TimeUnit.MINUTES.toMillis(1);
+ long cutoff5Min = now - TimeUnit.MINUTES.toMillis(5);
+ long cutoff30Min = now - TimeUnit.MINUTES.toMillis(30);
+
+ long invalidations10Sec = 0;
+ long invalidations1Min = 0;
+ long invalidations5Min = 0;
+ long invalidations30Min = 0;
+ Iterator<InvalidationBatch> it = batches.descendingIterator();
+ while (it.hasNext()) {
+ InvalidationBatch batch = it.next();
+ long timestamp = batch.timestampMillis();
+ if (timestamp < cutoff30Min) break;
+ int count = batch.count();
+ invalidations30Min += count;
+ if (timestamp >= cutoff5Min) invalidations5Min += count;
+ if (timestamp >= cutoff1Min) invalidations1Min += count;
+ if (timestamp >= cutoff10Sec) invalidations10Sec += count;
+ }
+ return new SlidingWindowCounts(
+ invalidations10Sec, invalidations1Min, invalidations5Min, invalidations30Min);
+ }
+
+ /**
+ * Snapshot of invalidation metrics from {@link CatalogdTableInvalidator}.
+ */
+ public record InvalidationMetrics(
+ long ttlInvalidations10Sec,
+ long ttlInvalidations1Min,
+ long ttlInvalidations5Min,
+ long ttlInvalidations30Min,
+ long memoryPressureInvalidations10Sec,
+ long memoryPressureInvalidations1Min,
+ long memoryPressureInvalidations5Min,
+ long memoryPressureInvalidations30Min,
+ long lastTtlInvalidationMillis,
+ long lastTtlInvalidatedTables,
+ long lastMemoryPressureInvalidationMillis,
+ long lastMemoryPressureInvalidatedTables) {}
+
+ /**
+ * Returns a consistent snapshot of all invalidation metrics. TTL and memory-pressure
+ * fields are read under separate locks since they are independent.
+ */
+ public InvalidationMetrics getMetrics() {
+ long ttlInvalidations10Sec;
+ long ttlInvalidations1Min;
+ long ttlInvalidations5Min;
+ long ttlInvalidations30Min;
+ long lastTtlInvalidationMillis;
+ long lastTtlInvalidatedTables;
+ synchronized (ttlInvalidationBatches_) {
+ SlidingWindowCounts ttlCounts =
+ countInvalidationsInSlidingWindows(ttlInvalidationBatches_);
+ ttlInvalidations10Sec = ttlCounts.invalidations10Sec();
+ ttlInvalidations1Min = ttlCounts.invalidations1Min();
+ ttlInvalidations5Min = ttlCounts.invalidations5Min();
+ ttlInvalidations30Min = ttlCounts.invalidations30Min();
+ InvalidationBatch lastTtl = lastTtlInvalidationBatch_;
+ lastTtlInvalidationMillis = lastTtl != null ? lastTtl.timestampMillis() : 0;
+ lastTtlInvalidatedTables = lastTtl != null ? lastTtl.count() : 0;
+ }
+
+ long memoryPressureInvalidations10Sec;
+ long memoryPressureInvalidations1Min;
+ long memoryPressureInvalidations5Min;
+ long memoryPressureInvalidations30Min;
+ long lastMemoryPressureInvalidationMillis;
+ long lastMemoryPressureInvalidatedTables;
+ synchronized (memoryPressureInvalidationBatches_) {
+ SlidingWindowCounts memoryCounts =
+ countInvalidationsInSlidingWindows(memoryPressureInvalidationBatches_);
+ memoryPressureInvalidations10Sec = memoryCounts.invalidations10Sec();
+ memoryPressureInvalidations1Min = memoryCounts.invalidations1Min();
+ memoryPressureInvalidations5Min = memoryCounts.invalidations5Min();
+ memoryPressureInvalidations30Min = memoryCounts.invalidations30Min();
+ InvalidationBatch lastMemory = lastMemoryPressureInvalidationBatch_;
+ lastMemoryPressureInvalidationMillis =
+ lastMemory != null ? lastMemory.timestampMillis() : 0;
+ lastMemoryPressureInvalidatedTables =
+ lastMemory != null ? lastMemory.count() : 0;
+ }
+
+ return new InvalidationMetrics(
+ ttlInvalidations10Sec,
+ ttlInvalidations1Min,
+ ttlInvalidations5Min,
+ ttlInvalidations30Min,
+ memoryPressureInvalidations10Sec,
+ memoryPressureInvalidations1Min,
+ memoryPressureInvalidations5Min,
+ memoryPressureInvalidations30Min,
+ lastTtlInvalidationMillis,
+ lastTtlInvalidatedTables,
+ lastMemoryPressureInvalidationMillis,
+ lastMemoryPressureInvalidatedTables);
+ }
+
private class DaemonThread implements Runnable {
@Override
public void run() {
diff --git a/fe/src/main/java/org/apache/impala/service/JniCatalog.java b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
index cfb2c6e68..c16b4f6c6 100644
--- a/fe/src/main/java/org/apache/impala/service/JniCatalog.java
+++ b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
@@ -39,6 +39,7 @@ import org.apache.impala.authorization.AuthorizationManager;
import org.apache.impala.catalog.Catalog;
import org.apache.impala.catalog.CatalogException;
import org.apache.impala.catalog.CatalogServiceCatalog;
+import org.apache.impala.catalog.CatalogdTableInvalidator;
import org.apache.impala.catalog.monitor.CatalogOperationTracker;
import org.apache.impala.catalog.Db;
import org.apache.impala.catalog.FeDb;
@@ -650,6 +651,37 @@ public class JniCatalog {
response.setCatalog_num_hms_clients_in_use(catalog_.getNumHmsClientsInUse());
response.setCatalog_num_loaded_tables(catalog_.getNumLoadedTables());
response.setEvent_metrics(catalog_.getEventProcessorMetrics());
+ response.setCatalog_num_ttl_invalidated_tables(
+ catalog_.getNumTtlInvalidatedTables());
+ response.setCatalog_num_memory_pressure_invalidated_tables(
+ catalog_.getNumMemoryPressureInvalidatedTables());
+
+ CatalogdTableInvalidator invalidator = catalog_.getCatalogdTableInvalidator();
+ if (invalidator != null) {
+ CatalogdTableInvalidator.InvalidationMetrics metrics =
+ invalidator.getMetrics();
+ response.setCatalog_ttl_invalidations_10s(metrics.ttlInvalidations10Sec());
+ response.setCatalog_ttl_invalidations_1m(metrics.ttlInvalidations1Min());
+ response.setCatalog_ttl_invalidations_5m(metrics.ttlInvalidations5Min());
+ response.setCatalog_ttl_invalidations_30m(metrics.ttlInvalidations30Min());
+ response.setCatalog_memory_pressure_invalidations_10s(
+ metrics.memoryPressureInvalidations10Sec());
+ response.setCatalog_memory_pressure_invalidations_1m(
+ metrics.memoryPressureInvalidations1Min());
+ response.setCatalog_memory_pressure_invalidations_5m(
+ metrics.memoryPressureInvalidations5Min());
+ response.setCatalog_memory_pressure_invalidations_30m(
+ metrics.memoryPressureInvalidations30Min());
+ response.setCatalog_last_ttl_invalidation_ms(
+ metrics.lastTtlInvalidationMillis());
+ response.setCatalog_last_ttl_invalidated_tables(
+ metrics.lastTtlInvalidatedTables());
+ response.setCatalog_last_memory_pressure_invalidation_ms(
+ metrics.lastMemoryPressureInvalidationMillis());
+ response.setCatalog_last_memory_pressure_invalidated_tables(
+ metrics.lastMemoryPressureInvalidatedTables());
+ }
+
return response;
});
}
diff --git a/fe/src/test/java/org/apache/impala/catalog/CatalogdTableInvalidatorTest.java b/fe/src/test/java/org/apache/impala/catalog/CatalogdTableInvalidatorTest.java
index 2a2c0e7b1..a947b0ebe 100644
--- a/fe/src/test/java/org/apache/impala/catalog/CatalogdTableInvalidatorTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/CatalogdTableInvalidatorTest.java
@@ -82,6 +82,93 @@ public class CatalogdTableInvalidatorTest {
Assert.assertFalse(catalog_.getTable(dbName, tblName).isLoaded());
}
+ /**
+ * TTL invalidation metrics: totals, sliding windows (staggered batches), last-batch
+ * fields, and isolation from memory-pressure counters.
+ * Only functional.alltypes is loaded first; after (ttlSec+10)s the daemon invalidates
+ * it (first batch, one table). Then functional.alltypesagg is loaded; after another
+ * (ttlSec+10)s the second batch runs. The two batches are spaced >10s apart in
+ * wall-clock time so the 10s sliding window counts only the latest batch (1 table)
+ * while 1m, 5m, and 30m include both (2 tables).
+ * Memory-pressure counters stay unchanged throughout.
+ */
+ @Test
+ public void testInvalidationMetrics() throws CatalogException, InterruptedException {
+ Reference<Boolean> tblWasRemoved = new Reference<>();
+ Reference<Boolean> dbWasAdded = new Reference<>();
+ String dbName = "functional";
+ String tblName1 = "alltypes";
+ String tblName2 = "alltypesagg";
+
+ catalog_.invalidateTable(new TTableName(dbName, tblName1), tblWasRemoved, dbWasAdded,
+ NoOpEventSequence.INSTANCE);
+ catalog_.invalidateTable(new TTableName(dbName, tblName2), tblWasRemoved, dbWasAdded,
+ NoOpEventSequence.INSTANCE);
+
+ long initialTtlCount = catalog_.getNumTtlInvalidatedTables();
+ long initialMemoryCount = catalog_.getNumMemoryPressureInvalidatedTables();
+
+ MockTicker ticker = new MockTicker();
+ CatalogdTableInvalidator.TIME_SOURCE = ticker;
+ // 1s TTL: invalidation batches are spaced >10s apart in wall-clock time so the 10s
+ // window sees only the latest batch while longer windows include both.
+ final long ttlSec = 1;
+ catalog_.setCatalogdTableInvalidator(
+ new CatalogdTableInvalidator(catalog_, /*unusedTableTtlSec=*/ttlSec,
+ /*invalidateTablesOnMemoryPressure=*/false, /*oldGenFullThreshold=*/0.6,
+ /*gcInvalidationFraction=*/0.1));
+
+ final long baseNanos = TimeUnit.HOURS.toNanos(1);
+ ticker.set(baseNanos);
+
+ Table table1 = catalog_.getOrLoadTable(dbName, tblName1, "test", null);
+ Assert.assertTrue(table1.isLoaded());
+
+ long previousTriggerCount = catalog_.getCatalogdTableInvalidator().scanCount_.get();
+ ticker.set(baseNanos + TimeUnit.SECONDS.toNanos(ttlSec + 10));
+ previousTriggerCount = waitForTrigger(previousTriggerCount);
+ Assert.assertFalse(catalog_.getTable(dbName, tblName1).isLoaded());
+
+ // Sliding-window metrics use wall-clock time; wait so the first batch falls outside
+ // the 10s window before the second batch runs.
+ sleep(11000);
+
+ ticker.set(baseNanos + TimeUnit.SECONDS.toNanos(ttlSec + 11));
+ previousTriggerCount = waitForTrigger(previousTriggerCount);
+ Table table2 = catalog_.getOrLoadTable(dbName, tblName2, "test", null);
+ Assert.assertTrue(table2.isLoaded());
+
+ ticker.set(baseNanos + TimeUnit.SECONDS.toNanos(2 * ttlSec + 21));
+ waitForTrigger(previousTriggerCount);
+ Assert.assertFalse(catalog_.getTable(dbName, tblName2).isLoaded());
+
+ Assert.assertEquals("TTL invalidation count should increase by 2",
+ initialTtlCount + 2, catalog_.getNumTtlInvalidatedTables());
+ Assert.assertEquals("Memory pressure invalidation count should not change",
+ initialMemoryCount, catalog_.getNumMemoryPressureInvalidatedTables());
+
+ CatalogdTableInvalidator.InvalidationMetrics metrics =
+ catalog_.getCatalogdTableInvalidator().getMetrics();
+
+ Assert.assertEquals("10-sec window should count only the latest batch", 1,
+ metrics.ttlInvalidations10Sec());
+ Assert.assertEquals("1-min window should include both batches", 2,
+ metrics.ttlInvalidations1Min());
+ Assert.assertEquals("5-min window should include both batches", 2,
+ metrics.ttlInvalidations5Min());
+ Assert.assertEquals("30-min window should include both batches", 2,
+ metrics.ttlInvalidations30Min());
+ Assert.assertTrue("Longer windows should be at least as large as shorter ones",
+ metrics.ttlInvalidations30Min() >= metrics.ttlInvalidations5Min()
+ && metrics.ttlInvalidations5Min() >= metrics.ttlInvalidations1Min()
+ && metrics.ttlInvalidations1Min() >= metrics.ttlInvalidations10Sec());
+
+ Assert.assertEquals("Last TTL batch should report 1 table", 1,
+ metrics.lastTtlInvalidatedTables());
+ Assert.assertTrue("Last TTL batch timestamp should be set",
+ metrics.lastTtlInvalidationMillis() > 0);
+ }
+
@After
public void cleanUp() {
catalog_.getCatalogdTableInvalidator().stop();
diff --git a/tests/custom_cluster/test_automatic_invalidation.py b/tests/custom_cluster/test_automatic_invalidation.py
index d123a6375..f94252ba5 100644
--- a/tests/custom_cluster/test_automatic_invalidation.py
+++ b/tests/custom_cluster/test_automatic_invalidation.py
@@ -43,6 +43,8 @@ class TestAutomaticCatalogInvalidation(CustomClusterTestSuite):
timeout = 20 if ImpalaTestClusterProperties.get_instance().runs_slowly() or\
(not IS_HDFS and not IS_LOCAL) else 10
timeout_flag = "--invalidate_tables_timeout_s=" + str(timeout)
+ metrics_test_ttl_s = 1
+ metrics_test_timeout_flag = ("--invalidate_tables_timeout_s=" + str(metrics_test_ttl_s))
def _get_catalog_object(self):
""" Return the catalog object of functional.alltypes serialized to string. """
@@ -211,3 +213,202 @@ class TestAutomaticCatalogInvalidation(CustomClusterTestSuite):
# Verify that the table metadata was actually invalidated
assert self.metadata_cache_string not in self._get_catalog_object(), \
"Table metadata should be invalidated after timeout"
+
+ @SkipIfFS.hive
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ catalogd_args=metrics_test_timeout_flag,
+ impalad_args=metrics_test_timeout_flag)
+ def test_invalidation_metrics(self):
+ """Test catalog invalidation metrics track TTL and memory pressure
+ based invalidations correctly."""
+
+ ttl_metric = "catalog.num-ttl-invalidated-tables"
+ memory_metric = "catalog.num-memory-pressure-invalidated-tables"
+ ttl_10s = "catalog.ttl-invalidations-10s"
+ ttl_1m = "catalog.ttl-invalidations-01m"
+ ttl_5m = "catalog.ttl-invalidations-05m"
+ ttl_30m = "catalog.ttl-invalidations-30m"
+ loaded_tables_metric = "catalog.num-loaded-tables"
+ last_ttl_ms = "catalog.last-ttl-invalidation-ms"
+ last_ttl_tables = "catalog.last-ttl-invalidated-tables"
+ last_mem_ms = "catalog.last-memory-pressure-invalidation-ms"
+ last_mem_tables = "catalog.last-memory-pressure-invalidated-tables"
+ catalogd = self.cluster.catalogd.service
+
+ # Clear any loaded tables from previous tests to start with a clean slate
+ self.execute_query("INVALIDATE METADATA")
+ catalogd.wait_for_metric_value(loaded_tables_metric, 0, timeout=5)
+
+ # Capture baseline counters - these are cumulative across test runs
+ baseline_ttl_count = catalogd.get_metric_value(ttl_metric)
+ baseline_memory_count = catalogd.get_metric_value(memory_metric)
+ assert baseline_ttl_count == 0, "TTL metric should be 0"
+ assert baseline_memory_count == 0, "Memory pressure metric should be 0"
+
+ # Total max wait = 15s (report delay) + TTL + 2s (buffer)
+ max_wait_time = 15 + self.metrics_test_ttl_s + 2
+
+ # Test 1: Load a single table and verify it gets invalidated
+ self.execute_query(self.query)
+ catalogd.wait_for_metric_value(loaded_tables_metric, 1, timeout=5)
+
+ # Wait for table to be unloaded
+ catalogd.wait_for_metric_value(loaded_tables_metric, 0, timeout=max_wait_time,
+ interval=2)
+
+ catalogd.wait_for_metric_value(ttl_metric, 1, timeout=5, allow_greater=True)
+ ttl_count_after_first = catalogd.get_metric_value(ttl_metric)
+ assert ttl_count_after_first == 1, ("TTL invalidation metric should be 1 (got %d)"
+ % ttl_count_after_first)
+
+ last_ttl_ms_1 = catalogd.get_metric_value(last_ttl_ms)
+ last_ttl_tables_1 = catalogd.get_metric_value(last_ttl_tables)
+ assert last_ttl_tables_1 == 1, (
+ "last-ttl-invalidated-tables should be 1 after first batch (got %d)"
+ % last_ttl_tables_1)
+ assert last_ttl_ms_1 > 0, (
+ "last-ttl-invalidation-ms should be set after first batch (got %d)"
+ % last_ttl_ms_1)
+ assert catalogd.get_metric_value(last_mem_ms) == 0, (
+ "last-memory-pressure-invalidation-ms should stay 0 (no memory-pressure batch)")
+ assert catalogd.get_metric_value(last_mem_tables) == 0, (
+ "last-memory-pressure-invalidated-tables should stay 0")
+
+ # Test 2: Verify memory pressure metric remained unchanged
+ memory_count_after_first = catalogd.get_metric_value(memory_metric)
+ assert memory_count_after_first == baseline_memory_count, (
+ "Memory pressure invalidation metric should remain %d (got %d)"
+ % (baseline_memory_count, memory_count_after_first))
+
+ # Test 3: Load multiple tables in one query
+ join_query = (
+ "SELECT 1 FROM functional.alltypessmall s "
+ "INNER JOIN functional.alltypesagg a ON s.id = a.id "
+ "INNER JOIN functional.alltypes t ON t.id = s.id LIMIT 1")
+ self.execute_query(join_query)
+ catalogd.wait_for_metric_value(loaded_tables_metric, 3, timeout=5)
+
+ # Wait for all join tables to be unloaded
+ catalogd.wait_for_metric_value(loaded_tables_metric, 0, timeout=max_wait_time,
+ interval=2)
+
+ # Wait for TTL metric to reach 4 (1 from first test + 3 from join)
+ expected_final_ttl = 4
+ catalogd.wait_for_metric_value(ttl_metric, expected_final_ttl,
+ timeout=5, allow_greater=True)
+ final_ttl_count = catalogd.get_metric_value(ttl_metric)
+ assert final_ttl_count == expected_final_ttl, ("TTL invalidation metric should be "
+ "%d (got %d)" % (expected_final_ttl, final_ttl_count))
+
+ last_ttl_ms_2 = catalogd.get_metric_value(last_ttl_ms)
+ last_ttl_tables_2 = catalogd.get_metric_value(last_ttl_tables)
+ # last-ttl-invalidated-tables is the size of the most recent batch only. With a
+ # short TTL, tables may expire across multiple daemon cycles (e.g. on multi-node
+ # clusters where last-used times are refreshed at slightly different times).
+ assert 1 <= last_ttl_tables_2 <= 3, (
+ "last-ttl-invalidated-tables after join round should be 1-3 (got %d)"
+ % last_ttl_tables_2)
+ assert last_ttl_ms_2 >= last_ttl_ms_1, (
+ "last-ttl-invalidation-ms should not go backwards (%d -> %d)"
+ % (last_ttl_ms_1, last_ttl_ms_2))
+ assert catalogd.get_metric_value(last_mem_ms) == 0, (
+ "last-memory-pressure-invalidation-ms should still be 0")
+ assert catalogd.get_metric_value(last_mem_tables) == 0, (
+ "last-memory-pressure-invalidated-tables should still be 0")
+
+ # Test 4: Verify memory pressure metric still unchanged
+ final_memory_count = catalogd.get_metric_value(memory_metric)
+ assert final_memory_count == baseline_memory_count, (
+ "Memory pressure invalidation metric should remain %d (got %d)"
+ % (baseline_memory_count, final_memory_count))
+
+ # Test 5: Verify all sliding window counts are accessible and non-negative
+ ttl_count_10s = catalogd.get_metric_value(ttl_10s)
+ ttl_count_1m = catalogd.get_metric_value(ttl_1m)
+ ttl_count_5m = catalogd.get_metric_value(ttl_5m)
+ ttl_count_30m = catalogd.get_metric_value(ttl_30m)
+
+ assert ttl_count_10s >= 0, (
+ "TTL 10-sec count should be non-negative (got %d)" % ttl_count_10s)
+
+ # Test 6: Verify window counts make sense (longer windows should contain at least
+ # as many invalidations as shorter windows)
+ assert ttl_count_1m >= ttl_count_10s, (
+ "1-min count (%d) should be >= 10-sec count (%d)"
+ % (ttl_count_1m, ttl_count_10s))
+ assert ttl_count_5m >= ttl_count_1m, (
+ "5-min count (%d) should be >= 1-min count (%d)" % (ttl_count_5m, ttl_count_1m))
+ assert ttl_count_30m >= ttl_count_5m, (
+ "30-min count (%d) should be >= 5-min count (%d)" % (ttl_count_30m, ttl_count_5m))
+
+ # Test 7: Window counts should reflect the 4 invalidations we triggered
+ assert ttl_count_30m == 4, (
+ "30-min window should contain exactly the 4 invalidations we triggered (got %d)"
+ % ttl_count_30m)
+
+ @SkipIfFS.hive
+ @pytest.mark.execute_serially
+ @CustomClusterTestSuite.with_args(
+ catalogd_args="--invalidate_tables_on_memory_pressure "
+ "--invalidate_tables_gc_old_gen_full_threshold=0 "
+ "--invalidate_tables_fraction_on_memory_pressure=1 "
+ "--invalidate_tables_timeout_s=0",
+ impalad_args="--invalidate_tables_on_memory_pressure "
+ "--invalidate_tables_timeout_s=0")
+ def test_memory_pressure_metrics(self):
+ """Test memory pressure invalidation metrics work correctly."""
+
+ ttl_metric = "catalog.num-ttl-invalidated-tables"
+ memory_metric = "catalog.num-memory-pressure-invalidated-tables"
+ memory_1m = "catalog.memory-pressure-invalidations-01m"
+ memory_5m = "catalog.memory-pressure-invalidations-05m"
+ memory_30m = "catalog.memory-pressure-invalidations-30m"
+ loaded_tables_metric = "catalog.num-loaded-tables"
+ last_mem_ms = "catalog.last-memory-pressure-invalidation-ms"
+ last_mem_tables = "catalog.last-memory-pressure-invalidated-tables"
+ catalogd = self.cluster.catalogd.service
+
+ # Clear state and get baseline
+ self.execute_query("INVALIDATE METADATA")
+ catalogd.wait_for_metric_value(loaded_tables_metric, 0, timeout=5)
+
+ baseline_ttl_count = catalogd.get_metric_value(ttl_metric)
+
+ self.execute_query(self.query)
+ catalogd.wait_for_metric_value(loaded_tables_metric, 1, timeout=5)
+
+ # Trigger memory pressure invalidation using GC
+ call(["jmap", "-histo:live", str(self.cluster.catalogd.get_pid())])
+
+ catalogd.wait_for_metric_value(loaded_tables_metric, 0, timeout=15, interval=2)
+
+ # Verify memory pressure metric increased
+ catalogd.wait_for_metric_value(memory_metric, 1, timeout=5, allow_greater=True)
+
+ assert catalogd.get_metric_value(last_mem_ms) > 0, (
+ "last-memory-pressure-invalidation-ms should be set after eviction (got %d)"
+ % catalogd.get_metric_value(last_mem_ms))
+ assert catalogd.get_metric_value(last_mem_tables) >= 1, (
+ "last-memory-pressure-invalidated-tables should be at least 1 (got %d)"
+ % catalogd.get_metric_value(last_mem_tables))
+
+ # Verify TTL metric unchanged
+ final_ttl_count = catalogd.get_metric_value(ttl_metric)
+ assert final_ttl_count == baseline_ttl_count, (
+ "TTL metric should remain %d (got %d)"
+ % (baseline_ttl_count, final_ttl_count))
+
+ # Sliding windows for the single memory-pressure batch we triggered
+ memory_count_1m = catalogd.get_metric_value(memory_1m)
+ memory_count_5m = catalogd.get_metric_value(memory_5m)
+ memory_count_30m = catalogd.get_metric_value(memory_30m)
+ assert memory_count_1m == 1, (
+ "1-min memory pressure window should show 1 invalidation (got %d)"
+ % memory_count_1m)
+ assert memory_count_5m == 1, (
+ "5-min memory pressure window should show 1 invalidation (got %d)"
+ % memory_count_5m)
+ assert memory_count_30m == 1, (
+ "30-min memory pressure window should show 1 invalidation (got %d)"
+ % memory_count_30m)