You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2026/08/21 11:20:09 UTC
(impala) 02/04: IMPALA-15178: Calcite Planner: support Iceberg time travel feature.
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 68ff2b20c2d61fcaa5a2ccd5b7f9b659870627ca
Author: Steve Carlin <sc...@cloudera.com>
AuthorDate: Wed Jul 8 13:55:43 2026 -0700
IMPALA-15178: Calcite Planner: support Iceberg time travel feature.
This commit supports the Iceberg time travel feature for queries.
Iceberg uses either the "FOR SYSTEM_TIME AS OF" or "FOR
SYSTEM_VERSION AS OF" sql phrase to retrieve data (with its proper
schema) at a given time.
Calcite does provide some code to support snapshots with its
SqlSnapshotNode (validation time) and LogicalSnapshot (optimization
time) objects. However, the Calcite implementation did not quite
fit the architecture used by Impala.
The SqlSnapshot object has a timestamp type "Period" and directly
creates a LogicalSnapshot RelNode on top of a LogicalTableScan. The
reason this didn't work for Impala is because:
- Iceberg works off a long value of snapshotId. While a timestamp
can be provided at query time, sometimes it is the snapshotId
itself that is provided through the SYSTEM_VERSION feature.
- Impala creates a pseudo-catalog table at analysis time called
IcebergTimeTravelTable. This table holds the schema at the given
time instance for the table. The problem for Calcite is that this
schema needs to be known at validation time to handle validation
steps like expanding a select star operation. When Calcite calls
back into Impala, it needs a unique table name to fetch this.
Some creative engineering was needed to handled these issues. The
architecture here is most likely safe from Calcite upgrades.
Highlights of these changes include:
- In the Parser.jj file, the Impala TimeTravelSpec object is created.
A JavaCCParserUtils utility class helps grab the special token after
the "AS OF" keyword. The Impala CUP parser has the code which parses
the time travel SQL. So some code was added to the CUP parser which
allows the object to be created, located in the ".cup" file and the
Parser java file.
- The SqlSnapshot Calcite class was extended (ImpalaSnapshotSqlNode
to hold onto the Impala specific TimeTravelSpec. The Period variable
is ignored. The tableRef name contains the table name concatenated
with "_tt_<TimeTravelSpec hash code>" to generate a unique table name
that will be found in the CalciteCatalogReader in the validation
stage.
- The CalciteMetadataHandler.TableVisitor is still called after the
parsing and before the validation. It walks through the SqlNode tree.
It still tracks all the tables as before. When it encounters the
ImpalaSnapshotSqlNode, it keeps a map of a table with a list of
TimeTravelSpecs which need a catalog table object. For normal tables
without a TimeTravelSpec, the list will contain a "null" for the
TimeTravelSpec. Duplicates are ok here as will be discussed in the
next bullet point.
- While there may be duplicates, the TableName objects duplicates
are removed via the TableVisitor.getTableNames() by the caller of
the visitor. These unique table names eventually get into the
StatementMetadataLoader class.
- The CalciteMetadataHandler.populateCalciteSchema is a callback
from the Impala framework called by the StatementMetadataLoader
before the Calcite validation step. In this phase, the tables get
loaded from catalogd. At this point, the Calcite schema gets loaded
with all the catalogd objects. For time travel objects, as mentioned
earlier, an IcebergTimeTravelTable object gets created. There may
be some TimeTravel tables that are duplicates, but it is not worth
the coding headache to make them unique since they only exist at
analysis time. The unique time travel name uses the same TimeTravelSpec
hash code in the ImpalaSnapshotSqlNode object which is how Calcite
can fetch the correct schema.
- At Logical node creation time, a LogicalSnapshot node is created.
This is not needed by Impala since the LogicalTableScan object will
contain the CalciteIcebergTable which contains the
IcebergTimeTravelTable, so it is removed from the logical RelNode
tree via the RemoveSnapshotRule.
- The existing ScanNodeHelper interface has a new getTimeTravelSpec
method which allows the IcebergScanPlanner to fetch the
TimeTravelSpec.
Testing: This was run against the time_travel e2e test in
test_iceberg.py and manually checked to ensure that there was never
any fallback. In a later commit, there will be tests to ensure that
fallback does not happen.
Change-Id: I6ab3466d6c453f8e030763749dd64903b8c41264
Reviewed-on: http://gerrit.cloudera.org:8080/24577
Tested-by: Impala Public Jenkins <im...@cloudera.com>
Reviewed-by: Steve Carlin <sc...@cloudera.com>
---
common/cup/sql-parser.cup | 10 +-
.../java/org/apache/impala/analysis/Parser.java | 22 +++-
.../impala/catalog/IcebergTimeTravelTable.java | 4 +
.../apache/impala/planner/IcebergScanPlanner.java | 8 +-
.../org/apache/impala/planner/ScanNodeHelper.java | 3 +
.../apache/impala/planner/ScanNodeHelperImpl.java | 15 ++-
.../apache/impala/planner/SingleNodePlanner.java | 3 +-
.../org/apache/impala/analysis/ParserTest.java | 8 +-
.../src/main/codegen/templates/Parser.jj | 44 ++++++--
.../impala/calcite/rel/node/ImpalaHdfsScanRel.java | 24 ++++-
.../impala/calcite/rules/RemoveSnapshotRule.java | 56 ++++++++++
.../apache/impala/calcite/schema/CalciteDb.java | 22 +++-
.../impala/calcite/schema/CalciteIcebergTable.java | 50 ++++++++-
.../apache/impala/calcite/schema/CalciteTable.java | 8 +-
.../calcite/service/CalciteAnalysisDriver.java | 4 +-
.../calcite/service/CalciteMetadataHandler.java | 114 ++++++++++++++++-----
.../impala/calcite/service/CalciteOptimizer.java | 4 +-
.../calcite/service/CalciteParsedStatement.java | 9 +-
.../impala/calcite/util/JavaCCParserUtils.java | 98 ++++++++++++++++++
.../calcite/validate/ImpalaSnapshotSqlNode.java | 100 ++++++++++++++++++
.../impala/calcite/util/JavaCCParserUtilTest.java | 86 ++++++++++++++++
.../iceberg-scan-metrics-with-deletes.test | 12 +--
22 files changed, 632 insertions(+), 72 deletions(-)
diff --git a/common/cup/sql-parser.cup b/common/cup/sql-parser.cup
index c78f3ff27..265342324 100755
--- a/common/cup/sql-parser.cup
+++ b/common/cup/sql-parser.cup
@@ -385,7 +385,11 @@ nonterminal String ident_or_unreserved;
// options, column attributes, etc.
nonterminal String word;
nonterminal String opt_comma;
-nonterminal StatementBase stmt;
+// stmt is declared as an Object to allow multiple entry points by the parser. As of
+// this current commit, it is a StatementBase for all queries coming from the API, but
+// it can also be a TimeTravelSpec when used by the Calcite parser for the specific
+// "FOR SYSTEM_<TIME|VERSION> AS OF" phrase.
+nonterminal Object stmt;
// Single select statement.
nonterminal SelectStmt select_stmt;
// Single values statement.
@@ -832,6 +836,10 @@ stmt ::=
{: RESULT = convert; :}
| kill_query_stmt: kill_query
{: RESULT = kill_query; :}
+ // Calcite specific. The parser within Calcite relies on the CUP parser
+ // to handle the "FOR SYSTEM_<TIME|VERSION> AS OF <obj>" syntax.
+ | opt_asof: opt_asof_phrase
+ {: RESULT = opt_asof_phrase; :}
;
load_stmt ::=
diff --git a/fe/src/main/java/org/apache/impala/analysis/Parser.java b/fe/src/main/java/org/apache/impala/analysis/Parser.java
index 4c111c6ab..c5ca89f52 100644
--- a/fe/src/main/java/org/apache/impala/analysis/Parser.java
+++ b/fe/src/main/java/org/apache/impala/analysis/Parser.java
@@ -19,6 +19,7 @@ package org.apache.impala.analysis;
import java.io.StringReader;
+import org.apache.impala.analysis.TimeTravelSpec;
import org.apache.impala.common.AnalysisException;
import org.apache.impala.common.ParseException;
import org.apache.impala.thrift.TQueryOptions;
@@ -50,11 +51,30 @@ public class Parser {
*/
public static StatementBase parse(String query, TQueryOptions options)
throws AnalysisException {
+ Object obj = parseString(query, options);
+ if (!(obj instanceof StatementBase)) {
+ throw new ParseException("Not a valid query: " + query);
+ }
+ return (StatementBase) obj;
+ }
+
+ public static TimeTravelSpec parseTimeTravelSpec(String query)
+ throws AnalysisException {
+ Object obj = parseString(query, new TQueryOptions());
+ if (!(obj instanceof TimeTravelSpec)) {
+ throw new ParseException("Not a valid timestamp or snapshot_id for iceberg: " +
+ query);
+ }
+ return (TimeTravelSpec) obj;
+ }
+
+ private static Object parseString(String query, TQueryOptions options)
+ throws AnalysisException {
SqlScanner input = new SqlScanner(new StringReader(query));
SqlParser parser = new SqlParser(input);
parser.setQueryOptions(options);
try {
- return (StatementBase) parser.parse().value;
+ return parser.parse().value;
} catch (AnalysisException e) {
// Pass along any analysis exceptions as they are. Since AnalysisException
// is a kind of Exception, we have to special-case it here
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergTimeTravelTable.java b/fe/src/main/java/org/apache/impala/catalog/IcebergTimeTravelTable.java
index 8493edbc2..16e0cdd3a 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergTimeTravelTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergTimeTravelTable.java
@@ -95,6 +95,10 @@ public class IcebergTimeTravelTable
public FeIcebergTable getBase() { return base_; }
+ public TimeTravelSpec getTimeTravelSpec() {
+ return timeTravelSpec_;
+ }
+
/**
* Initialize the columns from the schema corresponding to the time travel
* specification.
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 c2c7fbb86..a66d281f1 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
@@ -193,7 +193,7 @@ public class IcebergScanPlanner {
initPushDownHint();
extractIcebergConjuncts();
- snapshotId_ = IcebergUtil.getSnapshotId(getIceTable(), tblRef_.getTimeTravelSpec());
+ snapshotId_ = IcebergUtil.getSnapshotId(getIceTable(), helper.getTimeTravelSpec());
}
private ScanMetricsResult getScanMetrics() {
@@ -243,7 +243,7 @@ public class IcebergScanPlanner {
*/
private boolean needIcebergForPlanning() {
return !impalaIcebergPredicateMapping_.isEmpty()
- || tblRef_.getTimeTravelSpec() != null;
+ || helper_.getTimeTravelSpec() != null;
}
private void setFileDescriptorsBasedOnFileStore() throws ImpalaException {
@@ -711,7 +711,7 @@ public class IcebergScanPlanner {
Preconditions.checkState(allEqualityFieldIds_.isEmpty());
Preconditions.checkState(equalityIdsToDeleteFiles_.isEmpty());
- TimeTravelSpec timeTravelSpec = tblRef_.getTimeTravelSpec();
+ TimeTravelSpec timeTravelSpec = helper_.getTimeTravelSpec();
IcebergContentFileStore fileStore = getIceTable().getContentFileStore();
// 'metricsReporter_' is filled when the try-with-resources releases the FileScanTask
@@ -927,7 +927,7 @@ public class IcebergScanPlanner {
return new Pair<>(iceFileDesc, true);
}
- if (tblRef_.getTimeTravelSpec() == null) {
+ if (helper_.getTimeTravelSpec() == null) {
// We should always find the data files in the cache when not doing time travel.
throw new ImpalaRuntimeException(String.format("Cannot find file: %s in" +
" Iceberg table %s (snapshot id: %d) It's possibly missing from storage." +
diff --git a/fe/src/main/java/org/apache/impala/planner/ScanNodeHelper.java b/fe/src/main/java/org/apache/impala/planner/ScanNodeHelper.java
index f301a13a2..99e289ba3 100644
--- a/fe/src/main/java/org/apache/impala/planner/ScanNodeHelper.java
+++ b/fe/src/main/java/org/apache/impala/planner/ScanNodeHelper.java
@@ -22,6 +22,7 @@ import java.util.List;
import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.SlotDescriptor;
+import org.apache.impala.analysis.TimeTravelSpec;
/**
* Shared scan planning helpers and flags for the physical planner.
@@ -31,4 +32,6 @@ public interface ScanNodeHelper {
SlotDescriptor getCountStarOptimizationDescriptor(ScanNode scanNode,
Analyzer analyzer, List<Expr> conjuncts);
+
+ TimeTravelSpec getTimeTravelSpec();
}
diff --git a/fe/src/main/java/org/apache/impala/planner/ScanNodeHelperImpl.java b/fe/src/main/java/org/apache/impala/planner/ScanNodeHelperImpl.java
index a08427095..88c8d5a78 100644
--- a/fe/src/main/java/org/apache/impala/planner/ScanNodeHelperImpl.java
+++ b/fe/src/main/java/org/apache/impala/planner/ScanNodeHelperImpl.java
@@ -25,6 +25,7 @@ import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.MultiAggregateInfo;
import org.apache.impala.analysis.SlotDescriptor;
+import org.apache.impala.analysis.TimeTravelSpec;
/**
* Default {@link ScanNodeHelper} for the classic Impala planner.
@@ -33,12 +34,19 @@ public final class ScanNodeHelperImpl implements ScanNodeHelper {
private final boolean isDistinctOnly_;
+ private final TimeTravelSpec timeTravelSpec_;
+
public ScanNodeHelperImpl() {
- this(null);
+ this(null, null);
}
public ScanNodeHelperImpl(MultiAggregateInfo aggInfo) {
+ this(aggInfo, null);
+ }
+
+ public ScanNodeHelperImpl(MultiAggregateInfo aggInfo, TimeTravelSpec timeTravelSpec) {
isDistinctOnly_ = aggInfo != null && aggInfo.hasAllDistinctAgg();
+ timeTravelSpec_ = timeTravelSpec;
}
@Override
@@ -60,4 +68,9 @@ public final class ScanNodeHelperImpl implements ScanNodeHelper {
Preconditions.checkState(conjuncts.isEmpty());
return scanNode.applyCountStarOptimization(analyzer);
}
+
+ @Override
+ public TimeTravelSpec getTimeTravelSpec() {
+ return timeTravelSpec_;
+ }
}
diff --git a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
index 440f65518..b922bc3a4 100644
--- a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
@@ -1905,7 +1905,8 @@ public class SingleNodePlanner implements SingleNodePlannerIntf {
if (table instanceof FeFsTable) {
if (table instanceof FeIcebergTable) {
IcebergScanPlanner icebergPlanner = new IcebergScanPlanner(analyzer, ctx_, tblRef,
- conjuncts, aggInfo, new ScanNodeHelperImpl(aggInfo));
+ conjuncts, aggInfo,
+ new ScanNodeHelperImpl(aggInfo, tblRef.getTimeTravelSpec()));
return icebergPlanner.createIcebergScanPlan();
}
return createHdfsScanPlan(tblRef, aggInfo, conjuncts, analyzer);
diff --git a/fe/src/test/java/org/apache/impala/analysis/ParserTest.java b/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
index a2517d559..297f1fcac 100644
--- a/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
+++ b/fe/src/test/java/org/apache/impala/analysis/ParserTest.java
@@ -3821,9 +3821,9 @@ public class ParserTest extends FrontendTestBase {
"^\n" +
"Encountered: IDENTIFIER\n" +
"Expected: ALTER, COMMENT, COMPUTE, COPY, CREATE, DELETE, DESCRIBE, DROP, " +
- "EXPLAIN, GRANT, INSERT, INVALIDATE, KILL, LOAD, MERGE, OPTIMIZE, REFRESH, " +
- "REVOKE, SELECT, SET, SHOW, TRUNCATE, UNSET, UPDATE, UPSERT, USE, VALUES, " +
- "WITH\n");
+ "EXPLAIN, FOR, GRANT, INSERT, INVALIDATE, KILL, LOAD, MERGE, OPTIMIZE, " +
+ "REFRESH, REVOKE, SELECT, SET, SHOW, TRUNCATE, UNSET, UPDATE, UPSERT, USE, " +
+ "VALUES, WITH\n");
// missing select list
ParserError("select from t",
@@ -4477,7 +4477,7 @@ public class ParserTest extends FrontendTestBase {
@Test
public void TestSemiColon() {
- ParserError(";", "Syntax error");
+ ParserError(";", "Not a valid query: ;");
ParsesOk("SELECT 1;");
ParsesOk(" SELECT 1 ; ");
ParsesOk(" SELECT 1 ; ");
diff --git a/java/calcite-planner/src/main/codegen/templates/Parser.jj b/java/calcite-planner/src/main/codegen/templates/Parser.jj
index 418c1a2f6..2cab17d05 100644
--- a/java/calcite-planner/src/main/codegen/templates/Parser.jj
+++ b/java/calcite-planner/src/main/codegen/templates/Parser.jj
@@ -121,12 +121,18 @@ import org.apache.calcite.util.Pair;
import org.apache.calcite.util.SourceStringReader;
import org.apache.calcite.util.Util;
import org.apache.calcite.util.trace.CalciteTrace;
+import org.apache.impala.analysis.Parser;
+import org.apache.impala.analysis.SelectStmt;
+import org.apache.impala.analysis.TimeTravelSpec;
import org.apache.impala.analysis.TypesUtil;
import org.apache.impala.calcite.type.ImpalaSqlIntervalQualifier;
+import org.apache.impala.calcite.util.JavaCCParserUtils;
+import org.apache.impala.calcite.validate.ImpalaSnapshotSqlNode;
import org.apache.impala.calcite.operators.ImpalaCustomOperatorTable;
import org.apache.impala.calcite.operators.ImpalaConcatOrOperator;
import org.apache.impala.calcite.operators.ImpalaSqlNumericLiteral;
import org.apache.impala.catalog.ScalarType;
+import org.apache.impala.common.AnalysisException;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
@@ -2836,25 +2842,42 @@ JAVACODE SqlNode addLateral(SqlNode tableRef, boolean lateral) {
/**
* Parses a FOR SYSTEM_TIME clause following a table expression.
+ *
+ * Added code for Impala to handle FOR SYSTEM_VERSION and to help with parsing.
*/
SqlSnapshot Snapshot(SqlNode tableRef) :
{
final Span s;
final SqlNode e;
+ String snapshotType;
}
{
- { s = span(); } <FOR> <SYSTEM_TIME> <AS> <OF>
- // Syntax for temporal table in
- // standard SQL 2011 IWD 9075-2:201?(E) 7.6 <table reference>
- // supports grammar as following:
- // 1. datetime literal
- // 2. datetime value function, i.e. CURRENT_TIMESTAMP
- // 3. datetime term in 1 or 2 +(or -) interval term
+ { s = span(); }
+ (
+ LOOKAHEAD(4)
+ <FOR> <SYSTEM_TIME> <AS> <OF>
+ { snapshotType = "time";}
+ |
+ LOOKAHEAD(4)
+ <FOR> <SYSTEM_VERSION> <AS> <OF>
+ { snapshotType = "version";}
+ )
- // We extend to support column reference, use Expression
- // to simplify the parsing code.
e = Expression(ExprContext.ACCEPT_NON_QUERY) {
- return new SqlSnapshot(s.end(this), tableRef, e);
+ try {
+ SqlParserPos p = e.getParserPosition();
+ // Special utility used to grab the expression token.
+ String timeToken = JavaCCParserUtils.extractStringFromCoordinates(originalSql,
+ p.getLineNum(), p.getColumnNum(), p.getEndLineNum(), p.getEndColumnNum());
+ // Interesting logic here: An Impala TimeTravelSpec object is needed at
+ // validation time. So the Impala parser is invoked to create this
+ // TimeTravelSpec object here so that it can be used at validation time.
+ TimeTravelSpec tts = Parser.parseTimeTravelSpec("for system_" + snapshotType
+ + " as of " + timeToken);
+ return new ImpalaSnapshotSqlNode(s.end(this), tableRef, tts);
+ } catch (AnalysisException e2) {
+ throw new ParseException(e2.getMessage());
+ }
}
}
@@ -8466,6 +8489,7 @@ SqlPostfixOperator PostfixRowOperator() :
| < SYSTEM: "SYSTEM" >
| < SYSTEM_TIME: "SYSTEM_TIME" >
| < SYSTEM_USER: "SYSTEM_USER" >
+| < SYSTEM_VERSION: "SYSTEM_VERSION" >
| < TABLE: "TABLE" > { beforeTableName(); }
| < TABLE_NAME: "TABLE_NAME" >
| < TABLESAMPLE: "TABLESAMPLE" >
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaHdfsScanRel.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaHdfsScanRel.java
index c7ffc2369..7b5b49532 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaHdfsScanRel.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rel/node/ImpalaHdfsScanRel.java
@@ -28,6 +28,7 @@ import org.apache.impala.analysis.ExprSubstitutionMap;
import org.apache.impala.analysis.Path;
import org.apache.impala.analysis.SlotDescriptor;
import org.apache.impala.analysis.SlotRef;
+import org.apache.impala.analysis.TimeTravelSpec;
import org.apache.impala.analysis.TupleDescriptor;
import org.apache.impala.calcite.rel.phys.ImpalaHdfsScanNode;
import org.apache.impala.calcite.rel.util.ExprConjunctsConverter;
@@ -74,8 +75,12 @@ public class ImpalaHdfsScanRel extends TableScan
CalciteTable table = (CalciteTable) getTable();
- BaseTableRef baseTblRef =
- table.createBaseTableRef((SimplifiedAnalyzer) context.ctx_.getRootAnalyzer());
+ TimeTravelSpec timeTravelSpec = table instanceof CalciteIcebergTable
+ ? ((CalciteIcebergTable)table).getTimeTravelSpec()
+ : null;
+
+ BaseTableRef baseTblRef = table.createBaseTableRef(
+ (SimplifiedAnalyzer) context.ctx_.getRootAnalyzer(), timeTravelSpec);
produceSlotDescriptorsForTable(baseTblRef, context);
@@ -116,8 +121,8 @@ public class ImpalaHdfsScanRel extends TableScan
boolean isDistinctOnly = context.parentAggregate_ != null
&& context.parentAggregate_.hasDistinctOnly();
- CalcitePlanNodeHelper helper =
- new CalcitePlanNodeHelper(countStarDesc, isDistinctOnly);
+ CalcitePlanNodeHelper helper = new CalcitePlanNodeHelper(countStarDesc,
+ isDistinctOnly, timeTravelSpec);
PlanNode physicalNode;
if (SingleNodePlanner.addAcidSlotsIfNeeded(analyzer, baseTblRef,
@@ -378,9 +383,13 @@ public class ImpalaHdfsScanRel extends TableScan
private final SlotDescriptor countStarDesc_;
private final boolean isDistinctOnly_;
- CalcitePlanNodeHelper(SlotDescriptor countStarDesc, boolean isDistinctOnly) {
+ private final TimeTravelSpec timeTravelSpec_;
+
+ CalcitePlanNodeHelper(SlotDescriptor countStarDesc, boolean isDistinctOnly,
+ TimeTravelSpec timeTravelSpec) {
countStarDesc_ = countStarDesc;
isDistinctOnly_ = isDistinctOnly;
+ timeTravelSpec_ = timeTravelSpec;
}
@Override
@@ -393,5 +402,10 @@ public class ImpalaHdfsScanRel extends TableScan
Analyzer analyzer, List<Expr> conjuncts) {
return countStarDesc_;
}
+
+ @Override
+ public TimeTravelSpec getTimeTravelSpec() {
+ return timeTravelSpec_;
+ }
}
}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/rules/RemoveSnapshotRule.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rules/RemoveSnapshotRule.java
new file mode 100644
index 000000000..d3bf86586
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/rules/RemoveSnapshotRule.java
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.impala.calcite.rules;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Snapshot;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.impala.calcite.schema.CalciteIcebergTable;
+
+/**
+ * RemoveSnapshotRule removes the Snapshot RelNode. This is created by the
+ * SqlToRelNodeConverter because the SqlSnapshot SqlNode is created for
+ * Iceberg. However, the RelNode is not needed because the Iceberg snapshot
+ * table is resolved at validation time (as an IcebergTimeTravelTable for
+ * Iceberg) and exists in the LogicalTableScan.
+ */
+public class RemoveSnapshotRule extends RelOptRule {
+ public static final RemoveSnapshotRule INSTANCE =
+ new RemoveSnapshotRule();
+
+ private RemoveSnapshotRule() {
+ super(operand(Snapshot.class, operand(TableScan.class, none())));
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final Snapshot snapshot = call.rel(0);
+ final TableScan scan = call.rel(1);
+ // Only remove snapshot classes with an underlying CalciteIcebergTable
+ // which are handled in the analysis validation phase.
+ if (!(scan.getTable() instanceof CalciteIcebergTable)) {
+ return;
+ }
+
+ final RelNode input = snapshot.getInput();
+ call.transformTo(input);
+ }
+}
+
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteDb.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteDb.java
index 96ded22ae..9fbc655a6 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteDb.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteDb.java
@@ -27,9 +27,11 @@ import org.apache.calcite.schema.Table;
import org.apache.calcite.schema.impl.AbstractSchema;
import org.apache.calcite.schema.impl.ViewTable;
import org.apache.impala.analysis.Analyzer;
+import org.apache.impala.analysis.TimeTravelSpec;
import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.FeTable;
import org.apache.impala.catalog.FeView;
+import org.apache.impala.catalog.IcebergTimeTravelTable;
import org.apache.impala.catalog.HdfsTable;
import org.apache.impala.catalog.FeKuduTable;
import org.apache.impala.catalog.local.LocalFsTable;
@@ -61,6 +63,24 @@ public class CalciteDb extends AbstractSchema {
this.reader_ = reader;
}
+ /**
+ * Special table function for Iceberg time travel tables. Iceberg time travel table
+ * objects are created at analysis time (IcebergTimeTravelTable) wrapped around the
+ * real Iceberg catalog object (FeIcebergTable). The CalciteCatalogReader needs the
+ * time travel table object for validation items like "expand star".
+ */
+ public Builder addTimeTravelTable(String tableNameKey, TimeTravelSpec tts,
+ FeTable table, Analyzer analyzer) throws ImpalaException {
+
+ if (tableMap_.containsKey(tableNameKey)) return this;
+
+ FeIcebergTable feTimeTravelTable =
+ new IcebergTimeTravelTable((FeIcebergTable) table, tts);
+ tableMap_.put(tableNameKey,
+ new CalciteIcebergTable(feTimeTravelTable, reader_, analyzer, tableNameKey));
+ return this;
+ }
+
public Builder addTable(String tableName, FeTable feTable,
Analyzer analyzer) throws ImpalaException {
if (tableMap_.containsKey(tableName)) return this;
@@ -69,7 +89,7 @@ public class CalciteDb extends AbstractSchema {
if (table == null) {
throw new UnsupportedFeatureException("Table " + feTable.getFullName()
- + " has unsupported type " + table.getClass().getSimpleName() + ".");
+ + " has unsupported type " + feTable.getClass().getSimpleName() + ".");
}
tableMap_.put(tableName.toLowerCase(), table);
return this;
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteIcebergTable.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteIcebergTable.java
index 6b1c3cbb2..7467007c7 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteIcebergTable.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteIcebergTable.java
@@ -20,14 +20,18 @@ package org.apache.impala.calcite.schema;
import org.apache.calcite.prepare.CalciteCatalogReader;
import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.BaseTableRef;
+import org.apache.impala.analysis.TimeTravelSpec;
import org.apache.impala.catalog.Column;
import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.FeIcebergTable.Utils;
import org.apache.impala.catalog.IcebergColumn;
+import org.apache.impala.catalog.IcebergTimeTravelTable;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.util.IcebergUtil;
+import java.util.ArrayList;
import java.util.Collection;
+import java.util.List;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -44,9 +48,20 @@ public class CalciteIcebergTable extends CalciteTable {
// lazily populated
private Long recordCount_;
+ // Special name for the time travel object. It consists of the normal table
+ // name plus the "_tt_<hashcode of TimeTravelSpec>" string.
+ private final String timeTravelName_;
+
public CalciteIcebergTable(FeIcebergTable table, CalciteCatalogReader reader,
Analyzer analyzer) throws ImpalaException {
super(table, reader, analyzer);
+ timeTravelName_ = null;
+ }
+
+ public CalciteIcebergTable(FeIcebergTable table, CalciteCatalogReader reader,
+ Analyzer analyzer, String timeTravelName) throws ImpalaException {
+ super(table, reader, analyzer);
+ timeTravelName_ = timeTravelName;
}
public FeIcebergTable getFeIcebergTable() {
@@ -86,7 +101,8 @@ public class CalciteIcebergTable extends CalciteTable {
public boolean hasDeleteFiles() throws ImpalaException {
if (hasDeleteFiles_ == null) {
- hasDeleteFiles_ = FeIcebergTable.Utils.hasDeleteFiles(getFeIcebergTable(), null);
+ hasDeleteFiles_ = FeIcebergTable.Utils.hasDeleteFiles(getFeIcebergTable(),
+ getTimeTravelSpec());
}
return hasDeleteFiles_;
}
@@ -96,8 +112,9 @@ public class CalciteIcebergTable extends CalciteTable {
if (recordCount_ == null) {
FeIcebergTable feIcebergTable = getFeIcebergTable();
recordCount_ = hasDeleteFiles()
- ? Utils.getRecordCountV2(feIcebergTable, null)
- : Utils.getRecordCountV1(feIcebergTable.getIcebergApiTable(), null);
+ ? Utils.getRecordCountV2(feIcebergTable, getTimeTravelSpec())
+ : Utils.getRecordCountV1(feIcebergTable.getIcebergApiTable(),
+ getTimeTravelSpec());
}
return recordCount_;
}
@@ -113,4 +130,31 @@ public class CalciteIcebergTable extends CalciteTable {
}
}
+ @Override
+ public boolean isTemporal() {
+ return timeTravelName_ != null;
+ }
+
+ @Override
+ public List<String> getQualifiedName() {
+ List<String> qualifiedTableName = super.getQualifiedName();
+ if (!(getFeIcebergTable() instanceof IcebergTimeTravelTable)) {
+ return qualifiedTableName;
+ }
+ // all fields except the last are the same in the array.
+ List<String> newNames = new ArrayList<>();
+ for (int i = 0; i < qualifiedTableName.size() - 1; ++i) {
+ newNames.add(qualifiedTableName.get(i));
+ }
+ // last "table name" string contains the table name plus an appended
+ // "_tt_<hashcode of TimeTravelSpec>" string.
+ newNames.add(timeTravelName_.toUpperCase());
+ return newNames;
+ }
+
+ public TimeTravelSpec getTimeTravelSpec() {
+ return getFeIcebergTable() instanceof IcebergTimeTravelTable
+ ? ((IcebergTimeTravelTable) getFeIcebergTable()).getTimeTravelSpec()
+ : null;
+ }
}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteTable.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteTable.java
index 801e739d9..68087ba95 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteTable.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/schema/CalciteTable.java
@@ -40,6 +40,7 @@ import org.apache.impala.analysis.BaseTableRef;
import org.apache.impala.analysis.Expr;
import org.apache.impala.analysis.Path;
import org.apache.impala.analysis.TableRef;
+import org.apache.impala.analysis.TimeTravelSpec;
import org.apache.impala.analysis.TupleDescriptor;
import org.apache.impala.calcite.rel.util.ImpalaBaseTableRef;
import org.apache.impala.calcite.type.ImpalaTypeConverter;
@@ -125,12 +126,13 @@ public class CalciteTable extends RelOptAbstractTable
}
}
- public BaseTableRef createBaseTableRef(SimplifiedAnalyzer analyzer
- ) throws ImpalaException {
+ public BaseTableRef createBaseTableRef(SimplifiedAnalyzer analyzer,
+ TimeTravelSpec timeTravelSpec) throws ImpalaException {
TableRef tblRef = new TableRef(qualifiedTableName_, null);
- Path resolvedPath = analyzer.resolvePath(tblRef.getPath(), Path.PathType.TABLE_REF);
+ Path resolvedPath = analyzer.resolvePath(tblRef.getPath(), Path.PathType.TABLE_REF,
+ timeTravelSpec);
BaseTableRef baseTblRef = new ImpalaBaseTableRef(tblRef, resolvedPath, analyzer);
baseTblRef.analyze(analyzer);
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteAnalysisDriver.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteAnalysisDriver.java
index 3d6800c21..b5ce5e185 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteAnalysisDriver.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteAnalysisDriver.java
@@ -113,7 +113,7 @@ public class CalciteAnalysisDriver implements AnalysisDriver {
// Recall that parsedStmt_.getTablesInQuery(null) only contains TableName's in the
// given query but not the underlying tables referenced by a regular view.
CalciteMetadataHandler.populateCalciteSchema(reader_, ctx_.getCatalog(),
- stmtTableCache_, analyzer_);
+ stmtTableCache_, parsedStmt_.tableNameMap_, analyzer_);
typeFactory_ = ImpalaTypeFactoryImpl.INSTANCE;
sqlValidator_ = new ImpalaSqlValidatorImpl(
@@ -242,7 +242,7 @@ public class CalciteAnalysisDriver implements AnalysisDriver {
// Recurse if 'feTable' is also a view. Note that the privilege requests for the
// tables referenced by 'feTable' will be registered within the recursive call.
- registerPrivReqsInTables(tableVisitor.tableNames_,
+ registerPrivReqsInTables(tableVisitor.getTableNames(),
shouldMaskPrivChecks || childViewCreatedBySuperuser, catalog, validator);
// Set 'maskPrivChecks_' back to false in this case because we do not know if
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteMetadataHandler.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteMetadataHandler.java
index 86fb407db..f2f4a47fd 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteMetadataHandler.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteMetadataHandler.java
@@ -17,7 +17,6 @@
package org.apache.impala.calcite.service;
-import com.google.common.base.Splitter;
import org.apache.calcite.config.CalciteConnectionConfig;
import org.apache.calcite.config.CalciteConnectionConfigImpl;
import org.apache.calcite.config.CalciteConnectionProperty;
@@ -37,15 +36,22 @@ import org.apache.calcite.sql.util.SqlBasicVisitor;
import org.apache.impala.analysis.Analyzer;
import org.apache.impala.analysis.StmtMetadataLoader;
import org.apache.impala.analysis.TableName;
+import org.apache.impala.analysis.TimeTravelSpec;
import org.apache.impala.calcite.schema.CalciteDb;
import org.apache.impala.calcite.schema.ImpalaCalciteCatalogReader;
import org.apache.impala.calcite.type.ImpalaTypeFactoryImpl;
+import org.apache.impala.calcite.validate.ImpalaSnapshotSqlNode;
import org.apache.impala.catalog.FeCatalog;
import org.apache.impala.catalog.FeDb;
+import org.apache.impala.catalog.FeIcebergTable;
import org.apache.impala.catalog.FeTable;
+import org.apache.impala.common.AnalysisException;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.thrift.TQueryCtx;
+import com.google.common.base.Splitter;
+import com.google.common.collect.ImmutableSet;
+
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
@@ -93,6 +99,7 @@ public class CalciteMetadataHandler {
*/
public static List<String> populateCalciteSchema(CalciteCatalogReader reader,
FeCatalog catalog, StmtMetadataLoader.StmtTableCache stmtTableCache,
+ Map<TableName, List<TimeTravelSpec>> timeTravelSpecMap,
Analyzer analyzer) throws ImpalaException {
List<String> notFoundTables = new ArrayList<>();
CalciteSchema rootSchema = reader.getRootSchema();
@@ -116,7 +123,38 @@ public class CalciteMetadataHandler {
// first instance seen in the query.
CalciteDb.Builder dbBuilder =
dbSchemas.getOrDefault(tableName.getDb(), new CalciteDb.Builder(reader));
- dbBuilder.addTable(tableName.getTbl().toLowerCase(), feTable, analyzer);
+ String lowerCaseTableName = tableName.getTbl().toLowerCase();
+ List<TimeTravelSpec> timeTravelSpecs = timeTravelSpecMap.get(tableName);
+ // Special case: The tables in the for loop come from the tables loaded in the
+ // Impala stmtTableCache. For Kudu, this table can be different from the table
+ // name in the SQL query. In the e2e tests, this occurred on the
+ // functional_kudu.alltypesagg table which loaded in the
+ // functional_kudu.alltypesagg_idx table. The timeTravelSpecMap will not contain
+ // a value in this case, which is loaded in based on the user query. Since Kudu
+ // does not use time travel, we can safely load the table into the catalog without
+ // a time travel extension.
+ // If ever we do have a database that has time travel and manipulates the table
+ // name, this logic will have to change.
+ if (timeTravelSpecs == null) {
+ dbBuilder.addTable(lowerCaseTableName, feTable, analyzer);
+ } else {
+ for (TimeTravelSpec tts : timeTravelSpecs) {
+ // Actually, this is The normal case. Only Iceberg time travel tables will
+ // have a TimeTravelSpec
+ if (tts == null) {
+ dbBuilder.addTable(lowerCaseTableName, feTable, analyzer);
+ } else {
+ if (!(feTable instanceof FeIcebergTable)) {
+ throw new AnalysisException("Table '" + lowerCaseTableName +
+ "' is not a temporal table");
+ }
+ tts.analyze(analyzer);
+ String timeTravelTableKey = ImpalaSnapshotSqlNode.getIdentifierName(
+ lowerCaseTableName, tts);
+ dbBuilder.addTimeTravelTable(timeTravelTableKey, tts, feTable, analyzer);
+ }
+ }
+ }
dbSchemas.put(tableName.getDb().toLowerCase(), dbBuilder);
}
@@ -133,7 +171,7 @@ public class CalciteMetadataHandler {
*/
public static class TableVisitor extends SqlBasicVisitor<Void> {
private final String currentDb_;
- public final Set<TableName> tableNames_ = new HashSet<>();
+ private final Map<TableName, List<TimeTravelSpec>> tableNames_ = new HashMap<>();
// Error condition for now. Complex tables are not yet supported
// so if we see a table name that has more than 2 parts, this variable
@@ -149,6 +187,14 @@ public class CalciteMetadataHandler {
this.currentDb_ = currentDb.toLowerCase();
}
+ public Set<TableName> getTableNames() {
+ return ImmutableSet.copyOf(tableNames_.keySet());
+ }
+
+ public Map<TableName, List<TimeTravelSpec>> getTableNameMap() {
+ return tableNames_;
+ }
+
@Override
public Void visit(SqlCall call) {
if (call instanceof SqlWith) {
@@ -158,7 +204,7 @@ public class CalciteMetadataHandler {
if (call.getKind() == SqlKind.SELECT) {
SqlSelect select = (SqlSelect) call;
if (select.getFrom() != null) {
- tableNames_.addAll(getTableNames(select.getFrom()));
+ visitTableNameNode(select.getFrom());
}
}
@@ -182,43 +228,55 @@ public class CalciteMetadataHandler {
return v;
}
- private List<TableName> getTableNames(SqlNode fromNode) {
- List<TableName> localTableNames = new ArrayList<>();
+ private void extractTableName(SqlIdentifier identifer,
+ TimeTravelSpec timeTravelSpec) {
+ String tableNameString = identifer.toString();
+ List<String> parts = Splitter.on('.').splitToList(tableNameString);
+ if (parts.size() > 2) {
+ errorTables_.add(tableNameString);
+ return;
+ }
+ TableName tableName = parts.size() == 1
+ ? new TableName(currentDb_.toLowerCase(), parts.get(0).toLowerCase())
+ : new TableName(parts.get(0).toLowerCase(), parts.get(1).toLowerCase());
+
+ // Do not collect this table if 'tableNameToAdd' was already registered via
+ // a SqlWithItem node since in this case 'tableNameToAdd' is not an actual
+ // table.
+ if (parts.size() == 1 && isRegisteredBySqlWithItem(tableName)) {
+ return;
+ }
+
+ List<TimeTravelSpec> timeTravelSpecs =
+ tableNames_.getOrDefault(tableName, new ArrayList<>());
+ timeTravelSpecs.add(timeTravelSpec);
+ tableNames_.put(tableName, timeTravelSpecs);
+ }
+
+ private void visitTableNameNode(SqlNode fromNode) {
if (fromNode instanceof SqlIdentifier) {
- String tableName = fromNode.toString();
- List<String> parts = Splitter.on('.').splitToList(tableName);
- if (parts.size() == 1) {
- TableName tableNameToAdd = new TableName(
- currentDb_.toLowerCase(), parts.get(0).toLowerCase());
- // Do not collect this table if 'tableNameToAdd' was already registered via
- // a SqlWithItem node since in this case 'tableNameToAdd' is not an actual
- // table.
- if (!isRegisteredBySqlWithItem(tableNameToAdd)) {
- localTableNames.add(tableNameToAdd);
- }
- } else if (parts.size() == 2) {
- localTableNames.add(
- new TableName(parts.get(0).toLowerCase(), parts.get(1).toLowerCase()));
- } else {
- errorTables_.add(tableName);
- return localTableNames;
- }
+ extractTableName((SqlIdentifier) fromNode, null);
+ }
+
+ if (fromNode instanceof ImpalaSnapshotSqlNode) {
+ ImpalaSnapshotSqlNode snapshot = (ImpalaSnapshotSqlNode) fromNode;
+ extractTableName(snapshot.tableRefOriginal_, snapshot.timeTravelSpec_);
+ return;
}
// Join node has the tables in the left and right node.
if (fromNode instanceof SqlJoin) {
- localTableNames.addAll(getTableNames(((SqlJoin) fromNode).getLeft()));
- localTableNames.addAll(getTableNames(((SqlJoin) fromNode).getRight()));
+ visitTableNameNode(((SqlJoin) fromNode).getLeft());
+ visitTableNameNode(((SqlJoin) fromNode).getRight());
}
// Put references in the schema too
if (fromNode instanceof SqlBasicCall) {
SqlBasicCall basicCall = (SqlBasicCall) fromNode;
if (basicCall.getKind().equals(SqlKind.AS)) {
- localTableNames.addAll(getTableNames(basicCall.operand(0)));
+ visitTableNameNode(basicCall.operand(0));
}
}
- return localTableNames;
}
private boolean isRegisteredBySqlWithItem(TableName tableName) {
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteOptimizer.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteOptimizer.java
index 97f084331..32603f7f5 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteOptimizer.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteOptimizer.java
@@ -62,6 +62,7 @@ import org.apache.impala.calcite.rules.ImpalaProjectSimplifyRule;
import org.apache.impala.calcite.rules.ImpalaMQContext;
import org.apache.impala.calcite.rules.ImpalaRexExecutor;
import org.apache.impala.calcite.rules.RemoveInfrequentCTERule;
+import org.apache.impala.calcite.rules.RemoveSnapshotRule;
import org.apache.impala.calcite.schema.ImpalaCost;
import org.apache.impala.calcite.util.LogUtil;
import org.apache.impala.common.ImpalaException;
@@ -214,7 +215,8 @@ public class CalciteOptimizer implements CompilerStep {
ImpalaCoreRules.IMPALA_MINUS_TO_DISTINCT,
ImpalaCoreRules.COMBINE_VALUES_NODES,
ImpalaCoreRules.EXTRACT_LITERAL_AGG,
- ImpalaCoreRules.SORT_REMOVE_CONSTANT_KEYS
+ ImpalaCoreRules.SORT_REMOVE_CONSTANT_KEYS,
+ RemoveSnapshotRule.INSTANCE
));
builder.addMatchOrder(HepMatchOrder.BOTTOM_UP);
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteParsedStatement.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteParsedStatement.java
index 9d7c42a32..6138a5a53 100644
--- a/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteParsedStatement.java
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/service/CalciteParsedStatement.java
@@ -27,10 +27,13 @@ import org.apache.impala.analysis.AnalysisContext;
import org.apache.impala.analysis.ParsedStatement;
import org.apache.impala.analysis.TableName;
import org.apache.impala.analysis.StmtMetadataLoader;
+import org.apache.impala.analysis.TimeTravelSpec;
import org.apache.impala.analysis.ColumnLineageGraph.OperationType;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.thrift.TQueryCtx;
+import java.util.List;
+import java.util.Map;
/**
* Implemntation of ParsedStatement hook that holds the AST that
* is parsed from the sql String.
@@ -41,6 +44,8 @@ public class CalciteParsedStatement implements ParsedStatement {
private final String sql_;
private final Set<TableName> tableNames_;
+ public final Map<TableName, List<TimeTravelSpec>> tableNameMap_;
+
public CalciteParsedStatement(TQueryCtx queryCtx) throws ImpalaException {
sql_ = queryCtx.client_request.stmt;
CalciteQueryParser queryParser = new CalciteQueryParser(sql_);
@@ -56,7 +61,9 @@ public class CalciteParsedStatement implements ParsedStatement {
parsedNode_.accept(tableVisitor);
- tableNames_ = tableVisitor.tableNames_;
+ tableNames_ = tableVisitor.getTableNames();
+
+ tableNameMap_ = tableVisitor.getTableNameMap();
}
@Override
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/util/JavaCCParserUtils.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/util/JavaCCParserUtils.java
new file mode 100644
index 000000000..b6353656a
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/util/JavaCCParserUtils.java
@@ -0,0 +1,98 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.calcite.util;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.StringReader;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Utils to help with JavaCC.
+ *
+ * The code in this file was produced from a Gemini query.
+ */
+public class JavaCCParserUtils {
+
+ /**
+ * Extracts a string from a parser file based on line and column numbers.
+ *
+ * @param originalSource The full text of the parsed file.
+ * @param beginLine The 1-based starting line number.
+ * @param beginColumn The 1-based starting column number.
+ * @param endLine The 1-based ending line number.
+ * @param endColumn The 1-based ending column number.
+ * @return The extracted substring.
+ * @throws RuntimeException if the string could not be extracted.
+ */
+ public static String extractStringFromCoordinates(String originalSource,
+ int beginLine, int beginColumn, int endLine, int endColumn) {
+ try (BufferedReader reader = new BufferedReader(new StringReader(originalSource))) {
+ int currentLineNum = 1;
+ int currentAbsoluteIndex = 0;
+ int startIndex = -1;
+ int endIndex = -1;
+
+ String line;
+ String currentLine = originalSource;
+ while ((line = reader.readLine()) != null) {
+ // get the new current absolute index. On the first pass in the while
+ // loop, currentAbsoluteIndex will be at the beginning of the line. On
+ // subsequent calls, currentAbsoluteIndex will be pointed at the new
+ // line character or characters, so the "find" will skip over these
+ // characters and place at the beginning of the next line. Note, if
+ // there is a blank line in the middle, currentAbsoluteIndex will not
+ // change, but that's ok because the endToken will never be pointed
+ // to a new line character.
+ currentAbsoluteIndex = currentLine.indexOf(line, currentAbsoluteIndex);
+
+ // If we are on the starting line, calculate the starting character index
+ if (currentLineNum == beginLine) {
+ // Convert 1-based column to 0-based index
+ startIndex = currentAbsoluteIndex + (beginColumn - 1);
+ }
+
+ // If we are on the ending line, calculate the ending character index
+ if (currentLineNum == endLine) {
+ endIndex = currentAbsoluteIndex + endColumn;
+ // Last character should always be part of a token, so never a
+ // whitespace character.
+ Preconditions.checkState(
+ !Character.isWhitespace(originalSource.charAt(endIndex - 1)));
+ break; // We have both indices, we can stop reading
+ }
+
+ currentAbsoluteIndex += line.length();
+ currentLineNum++;
+ }
+
+ // Extract the string if valid coordinates were found
+ if (startIndex != -1 && endIndex != -1 && startIndex <= endIndex &&
+ endIndex <= originalSource.length()) {
+ return originalSource.substring(startIndex, endIndex);
+ }
+ } catch (IOException e) {
+ // This should never happen. Assuming the caller passes in correct values.
+ throw new RuntimeException(e);
+ }
+ throw new RuntimeException("Error in JavaCC parser, bad coordinates passed in." +
+ "beginLine=" + beginLine + ", beginColumn=" + beginColumn + ", endLine=" +
+ endLine + ",endColumn=" + endColumn);
+ }
+}
diff --git a/java/calcite-planner/src/main/java/org/apache/impala/calcite/validate/ImpalaSnapshotSqlNode.java b/java/calcite-planner/src/main/java/org/apache/impala/calcite/validate/ImpalaSnapshotSqlNode.java
new file mode 100644
index 000000000..228cf5fe5
--- /dev/null
+++ b/java/calcite-planner/src/main/java/org/apache/impala/calcite/validate/ImpalaSnapshotSqlNode.java
@@ -0,0 +1,100 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.calcite.validate;
+
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlSnapshot;
+import org.apache.calcite.sql.SqlTimestampLiteral;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.parser.SqlParserUtil;
+import org.apache.impala.analysis.TimeTravelSpec;
+
+import com.google.common.hash.Hashing;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The ImpalaSnapshotSqlNode extends the Calcite SnapshotSqlNode with a TimeTravelSpec
+ * needed by Impala for the snapshot.
+ */
+public class ImpalaSnapshotSqlNode extends SqlSnapshot {
+
+ private static final String DUMMY_TIMESTAMP = "1970-01-01 00:00:01";
+
+ public final TimeTravelSpec timeTravelSpec_;
+ public final SqlIdentifier tableRefOriginal_;
+
+ public ImpalaSnapshotSqlNode(SqlParserPos parserPos, SqlNode tableRefInput,
+ TimeTravelSpec t) {
+ super(parserPos, getTimeTravelSpecTableRef((SqlIdentifier) tableRefInput, t),
+ getDummyPeriod(parserPos));
+ tableRefOriginal_ = (SqlIdentifier) tableRefInput;
+ timeTravelSpec_ = t;
+ }
+
+ /**
+ * Create a new time travel SqlNode which contains a modified table name that
+ * contains the real Iceberg table concatenated with the unique identifier
+ * "_tt_<hash_code>".
+ * Note: It would be a little better to use the Impala snapshotId rather than the
+ * object's hashcode as a unique identifier. However, this object is created at
+ * parse time along with its unique identifier, and the catalog table object is
+ * not known at this point.
+ */
+ private static SqlNode getTimeTravelSpecTableRef(SqlIdentifier tableRef,
+ TimeTravelSpec t) {
+ List<String> names = new ArrayList<>();
+ // The first <n-1> names (probably just the db) do not change. The table name is
+ // the last element in the array.
+ for (int i = 0; i < tableRef.names.size() - 1; ++i) {
+ names.add(tableRef.names.get(i));
+ }
+ String newTableName =
+ getIdentifierName(tableRef.names.get(tableRef.names.size() - 1), t);
+ names.add(newTableName);
+ return new SqlIdentifier(names, tableRef.getParserPosition());
+ }
+
+ /**
+ * Create a unique identifier name for this TimeTravelSpec.
+ * The reason this mechanism is used is because the name is created at Parser
+ * time, before the TimeTravelSpec is analyzed. This name will be also placed
+ * in the CalciteCatalogReader as the Iceberg table name. At validation time,
+ * Calcite then uses the dummy table name to match up with the Snapshot SqlNode.
+ */
+ public static String getIdentifierName(String tableName, TimeTravelSpec tts) {
+ // Use this hash code which is stable across different compilations.
+ String hashCode = Hashing.murmur3_128()
+ .hashString(tts.toSql(), StandardCharsets.UTF_8)
+ .toString().substring(0,15);
+ return tableName.toLowerCase() + "_tt_" + hashCode;
+ }
+
+ /**
+ * The "period" variable is not used by Impala since Impala cannot just
+ * use a timestamp field. The TimeTravelSpec can contain not only a timestamp,
+ * but also a long snapshotId. A null is not allowed, so a random dummy timestamp
+ * is provided.
+ */
+ public static SqlTimestampLiteral getDummyPeriod(SqlParserPos pos) {
+ return SqlParserUtil.parseTimestampLiteral(DUMMY_TIMESTAMP, pos);
+ }
+}
diff --git a/java/calcite-planner/src/test/java/org/apache/impala/calcite/util/JavaCCParserUtilTest.java b/java/calcite-planner/src/test/java/org/apache/impala/calcite/util/JavaCCParserUtilTest.java
new file mode 100644
index 000000000..a88853c56
--- /dev/null
+++ b/java/calcite-planner/src/test/java/org/apache/impala/calcite/util/JavaCCParserUtilTest.java
@@ -0,0 +1,86 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.impala.calcite.util;
+
+import org.junit.Test;
+import static org.junit.Assert.assertEquals;
+
+public class JavaCCParserUtilTest {
+ @Test
+ public void testWholeString() {
+ String s = "test token";
+
+ String newString = JavaCCParserUtils.extractStringFromCoordinates(
+ s, 1, 1, 1, 10);
+ assertEquals(newString, "test token");
+ }
+
+ @Test
+ public void testLastCharFirstLine() {
+ String s = "test token";
+
+ String newString = JavaCCParserUtils.extractStringFromCoordinates(
+ s, 1, 10, 1, 10);
+ assertEquals(newString, "n");
+ }
+
+ @Test
+ public void testMultiline() {
+ String s = "test token\n next line";
+
+ String newString = JavaCCParserUtils.extractStringFromCoordinates(
+ s, 1, 1, 2, 10);
+ assertEquals(newString, "test token\n next line");
+ }
+
+ @Test
+ public void testMultilineCRLF() {
+ String s = "test token\r\n next line";
+
+ String newString = JavaCCParserUtils.extractStringFromCoordinates(
+ s, 1, 1, 2, 10);
+ assertEquals(newString, "test token\r\n next line");
+ }
+
+ @Test
+ public void testMultilineMiddle() {
+ String s = "first line\ntest token\n next line";
+
+ String newString = JavaCCParserUtils.extractStringFromCoordinates(
+ s, 2, 2, 3, 9);
+ assertEquals(newString, "est token\n next lin");
+ }
+
+ @Test
+ public void testMultilineMiddleSingleLine() {
+ String s = "first line\ntest token\n next line";
+
+ String newString = JavaCCParserUtils.extractStringFromCoordinates(
+ s, 2, 2, 2, 10);
+ assertEquals(newString, "est token");
+ }
+
+ @Test
+ public void testMultilineEmptyMiddleLine() {
+ String s = "first line\ntest token\n\n next line";
+
+ String newString = JavaCCParserUtils.extractStringFromCoordinates(
+ s, 2, 2, 4, 2);
+ assertEquals(newString, "est token\n\n n");
+ }
+}
diff --git a/testdata/workloads/functional-query/queries/QueryTest/iceberg-scan-metrics-with-deletes.test b/testdata/workloads/functional-query/queries/QueryTest/iceberg-scan-metrics-with-deletes.test
index 8b2c9eedd..fcec3fdaa 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/iceberg-scan-metrics-with-deletes.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/iceberg-scan-metrics-with-deletes.test
@@ -6,7 +6,7 @@ select * from
functional_parquet.iceberg_v2_positional_delete_all_rows FOR SYSTEM_VERSION AS OF POS_DELETE_ALL_ROWS_SNAPSHOT_ID,
functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files FOR SYSTEM_VERSION AS OF NOT_ALL_DATA_FILES_HAVE_DELETE_FILES_SNAPSHOT_ID
---- RUNTIME_PROFILE
-Iceberg Plan Metrics for Node 00:
+row_regex:Iceberg Plan Metrics for Node 0[05]:
row_regex: total-planning-duration:
result-data-files: 1
result-delete-files: 0
@@ -23,13 +23,13 @@ row_regex: total-delete-file-size-in-bytes: .+ \(\d+\)
indexed-delete-files: 0
equality-delete-files: 0
positional-delete-files: 0
-Iceberg Plan Metrics for Node 03:
+row_regex:Iceberg Plan Metrics for Node 0[37]:
result-delete-files: 1
total-delete-manifests: 1
scanned-delete-manifests: 1
indexed-delete-files: 1
positional-delete-files: 1
-Iceberg Plan Metrics for Node 08:
+row_regex:Iceberg Plan Metrics for Node [01][82]:
result-data-files: 4
result-delete-files: 2
total-data-manifests: 4
@@ -46,9 +46,9 @@ select * from
functional_parquet.iceberg_v2_positional_delete_all_rows FOR SYSTEM_VERSION AS OF POS_DELETE_ALL_ROWS_SNAPSHOT_ID,
functional_parquet.iceberg_v2_positional_not_all_data_files_have_delete_files FOR SYSTEM_VERSION AS OF NOT_ALL_DATA_FILES_HAVE_DELETE_FILES_SNAPSHOT_ID
---- RUNTIME_PROFILE
-Iceberg Plan Metrics for Node 00:
+row_regex:Iceberg Plan Metrics for Node 0[05]:
Planning done without Iceberg: no Iceberg scan metrics available.
-Iceberg Plan Metrics for Node 03:
+row_regex:Iceberg Plan Metrics for Node 0[37]:
row_regex: total-planning-duration:
result-data-files: 1
result-delete-files: 1
@@ -65,7 +65,7 @@ row_regex: total-delete-file-size-in-bytes: .+ \(\d+\)
indexed-delete-files: 1
equality-delete-files: 0
positional-delete-files: 1
-Iceberg Plan Metrics for Node 08:
+row_regex:Iceberg Plan Metrics for Node [01][82]:
result-data-files: 4
result-delete-files: 2
total-data-manifests: 4