morningman commented on code in PR #51272:
URL: https://github.com/apache/doris/pull/51272#discussion_r2119218963
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/TableScanParams.java:
##########
@@ -19,28 +19,45 @@
import com.google.common.collect.ImmutableMap;
+import java.util.List;
import java.util.Map;
public class TableScanParams {
public static String INCREMENTAL_READ = "incr";
+ public static String BRANCH = "branch";
+ public static String TAG = "tag";
private final String paramType;
- private final Map<String, String> params;
+ private final Map<String, String> mapParams;
Review Comment:
Add some comment for these fields, like what kind of params does these
fields saved, can use som examples
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java:
##########
@@ -157,6 +161,10 @@ public Integer initialValue() {
public static final String IDENTITY = "identity";
public static final int PARTITION_DATA_ID_START = 1000; //
org.apache.iceberg.PartitionSpec
+ private static final Pattern BRANCH = Pattern.compile("branch_(.*)");
Review Comment:
`BRANCH` and `TAG` is not used
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalHudiScan.java:
##########
@@ -62,8 +62,7 @@ public class LogicalHudiScan extends LogicalFileScan {
private static final Logger LOG =
LogManager.getLogger(LogicalHudiScan.class);
// for hudi incremental read
- private final Optional<TableScanParams> scanParams;
- private final Optional<IncrementalRelation> incrementalRelation;
+ private Optional<IncrementalRelation> incrementalRelation;
Review Comment:
In Nereids, most of all objects are required to be immutable.
So here we should retain the `final` qualifier. And create a new one if we
need to modify a field in class
##########
regression-test/suites/external_table_p0/iceberg/iceberg_query_tag_branch.groovy:
##########
@@ -0,0 +1,70 @@
+// 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.
+
+suite("iceberg_query_tag_branch",
"p0,external,doris,external_docker,external_docker_doris") {
+
+ String enabled = context.config.otherConfigs.get("enableIcebergTest")
+ if (enabled == null || !enabled.equalsIgnoreCase("true")) {
+ logger.info("disable iceberg test.")
+ return
+ }
+
+ String rest_port = context.config.otherConfigs.get("iceberg_rest_uri_port")
+ String minio_port = context.config.otherConfigs.get("iceberg_minio_port")
+ String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
+ String catalog_name = "iceberg_query_tag_branch"
+
+ sql """drop catalog if exists ${catalog_name}"""
+ sql """
+ CREATE CATALOG ${catalog_name} PROPERTIES (
+ 'type'='iceberg',
+ 'iceberg.catalog.type'='rest',
+ 'uri' = 'http://${externalEnvIp}:${rest_port}',
+ "s3.access_key" = "admin",
+ "s3.secret_key" = "password",
+ "s3.endpoint" = "http://${externalEnvIp}:${minio_port}";,
+ "s3.region" = "us-east-1"
+ );"""
+
+
+ logger.info("catalog " + catalog_name + " created")
+ sql """switch ${catalog_name};"""
+ logger.info("switched to catalog " + catalog_name)
+ sql """ use test_db;"""
+
+ qt_branch_1 """ select * from tag_branch_table@branch(b1) order by c1;"""
+ qt_branch_2 """ select * from tag_branch_table@branch(b2) order by c1
;"""
+ qt_branch_3 """ select * from tag_branch_table@branch(b3) order by c1
;"""
+ qt_branch_4 """ select * from tag_branch_table@branch('name'='b1') order
by c1 ;"""
+ qt_branch_5 """ select * from tag_branch_table@branch('name'='b2') order
by c1 ;"""
+ qt_branch_6 """ select * from tag_branch_table@branch('name'='b3') order
by c1 ;"""
+
+ qt_tag_1 """ select * from tag_branch_table@tag(t1) order by c1 ;"""
+ qt_tag_2 """ select * from tag_branch_table@tag(t2) order by c1 ;"""
+ qt_tag_3 """ select * from tag_branch_table@tag(t3) order by c1 ;"""
+ qt_tag_4 """ select * from tag_branch_table@tag('name'='t1') order by c1
;"""
+ qt_tag_5 """ select * from tag_branch_table@tag('name'='t2') order by c1
;"""
+ qt_tag_6 """ select * from tag_branch_table@tag('name'='t3') order by c1
;"""
+
+ qt_version_1 """ select * from tag_branch_table for version as of 'b1'
order by c1 ;"""
+ qt_version_2 """ select * from tag_branch_table for version as of 'b2'
order by c1 ;"""
+ qt_version_3 """ select * from tag_branch_table for version as of 'b3'
order by c1 ;"""
+ qt_version_4 """ select * from tag_branch_table for version as of 't1'
order by c1 ;"""
+ qt_version_5 """ select * from tag_branch_table for version as of 't2'
order by c1;"""
+ qt_version_6 """ select * from tag_branch_table for version as of 't3'
order by c1 ;"""
Review Comment:
Better add following cases:1
1. query branch/tag with "count(*) pushdown" feature.
2. query branch/tag with "batch mode" enabled.
3. query part of columns instead of `select *`
4. query branch/tag in subquery, join statement
5. query wrong branch/tag, or invalid parameters
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/FileScanNode.java:
##########
@@ -105,15 +105,21 @@ public String getNodeExplainString(String prefix,
TExplainLevel detailLevel) {
}
output.append(prefix);
- if (isBatchMode()) {
+ boolean isBatch;
Review Comment:
Is this a bug? If yes, better submit a separate PR to fix it
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUtils.java:
##########
@@ -782,18 +790,106 @@ public static CloseableIterable<ManifestFile>
getMatchingManifest(
return matchingManifests;
}
- // get snapshot id from query like 'for version/time as of'
- public static long getQuerySpecSnapshot(Table table, TableSnapshot
queryTableSnapshot) {
- TableSnapshot.VersionType type = queryTableSnapshot.getType();
+ // get snapshot id from query like 'for version/time as of' or
'@branch/@tag'
+ public static IcebergTableQueryInfo getQuerySpecSnapshot(
+ Table table,
+ Optional<TableSnapshot> queryTableSnapshot,
+ Optional<TableScanParams> scanParams) throws UserException {
+
+ Preconditions.checkArgument(
+ queryTableSnapshot.isPresent() ||
isIcebergBranchOrTag(scanParams),
+ "should spec version or time or branch or tag");
+
+ // not support `select * from tb@branch/tag(b) for version/time as of
...`
+ Preconditions.checkArgument(
+ !(queryTableSnapshot.isPresent() &&
isIcebergBranchOrTag(scanParams)),
+ "could not spec a version/time with tag/branch");
+
+ // solve @branch/@tag
+ if (scanParams.isPresent()) {
+ String refName;
+ TableScanParams params = scanParams.get();
+ if (!params.getMapParams().isEmpty()) {
+ refName = params.getMapParams().get("name");
+ } else {
+ refName = params.getListParams().get(0);
+ }
+ SnapshotRef snapshotRef = table.refs().get(refName);
+ if (params.isBranch()) {
+ if (snapshotRef == null || !snapshotRef.isBranch()) {
+ throw new UserException("Table " + table.name() + " does
not have branch named " + refName);
Review Comment:
Please add negative test cases for these invalid parameters.
##########
fe/fe-core/src/main/java/org/apache/doris/analysis/TableScanParams.java:
##########
@@ -19,28 +19,45 @@
import com.google.common.collect.ImmutableMap;
+import java.util.List;
import java.util.Map;
public class TableScanParams {
public static String INCREMENTAL_READ = "incr";
+ public static String BRANCH = "branch";
+ public static String TAG = "tag";
private final String paramType;
- private final Map<String, String> params;
+ private final Map<String, String> mapParams;
+ private final List<String> listParams;
- public TableScanParams(String paramType, Map<String, String> params) {
+ public TableScanParams(String paramType, Map<String, String> mapParams,
List<String> listParams) {
this.paramType = paramType;
- this.params = params == null ? ImmutableMap.of() :
ImmutableMap.copyOf(params);
+ this.mapParams = mapParams == null ? ImmutableMap.of() :
ImmutableMap.copyOf(mapParams);
+ this.listParams = listParams;
+ }
+
+ public List<String> getListParams() {
+ return listParams;
}
public String getParamType() {
return paramType;
}
- public Map<String, String> getParams() {
- return params;
+ public Map<String, String> getMapParams() {
+ return mapParams;
}
public boolean incrementalRead() {
return INCREMENTAL_READ.equals(paramType);
}
+
+ public boolean isBranch() {
+ return BRANCH.equals(paramType);
Review Comment:
Do we need to consider case sensibility?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
