[Enhance](Iceberg) Enhance branch/tag operations with validation and comprehensive test coverage

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java

@@ -455,6 +455,11 @@ public void createOrReplaceBranchImpl(ExternalTable dorisTable, CreateOrReplaceB
                             + ", error message is: {} " + ExceptionUtils.getRootCauseMessage(e), e);
         }
         String branchName = branchInfo.getBranchName();
+
+        if (branchName == null || branchName.trim().isEmpty()) {
+            throw new UserException("Branch name cannot be empty");
+        }
+
         boolean refExists = null != icebergTable.refs().get(branchName);
         boolean create = branchInfo.getCreate();
         boolean replace = branchInfo.getReplace();
@@ -535,6 +540,11 @@ public void createOrReplaceTagImpl(ExternalTable dorisTable, CreateOrReplaceTagI
         }
 
         String tagName = tagInfo.getTagName();
+
+        if (tagName == null || tagName.trim().isEmpty()) {
+            throw new UserException("Tag name cannot be empty");
+        }
+
         boolean create = tagInfo.getCreate();
         boolean replace = tagInfo.getReplace();
         boolean ifNotExists = tagInfo.getIfNotExists();

fe/fe-core/src/main/java/org/apache/doris/nereids/StatementContext.java

@@ -781,16 +781,19 @@ public void addPlannerHook(PlannerHook plannerHook) {
     }
 
     /**
-     * Load snapshot information of mvcc
+     * Load snapshot information of mvcc for a specific table.
+     *
+     * @param specificTable specific table to load snapshot for.
+     * @param tableSnapshot table snapshot info
+     * @param scanParams table scan params (e.g., branch/tag for Iceberg tables)
      */
-    public void loadSnapshots(Optional<TableSnapshot> tableSnapshot, Optional<TableScanParams> scanParams) {
-        for (TableIf tableIf : tables.values()) {
-            if (tableIf instanceof MvccTable) {
-                MvccTableInfo mvccTableInfo = new MvccTableInfo(tableIf);
-                // may be set by MTMV, we can not load again
-                if (!snapshots.containsKey(mvccTableInfo)) {
-                    snapshots.put(mvccTableInfo, ((MvccTable) tableIf).loadSnapshot(tableSnapshot, scanParams));
-                }
+    public void loadSnapshots(TableIf specificTable, Optional<TableSnapshot> tableSnapshot,
+            Optional<TableScanParams> scanParams) {
+        if (specificTable instanceof MvccTable) {
+            MvccTableInfo mvccTableInfo = new MvccTableInfo(specificTable);
+            if (!snapshots.containsKey(mvccTableInfo)) {
+                snapshots.put(mvccTableInfo,
+                        ((MvccTable) specificTable).loadSnapshot(tableSnapshot, scanParams));
             }
         }
     }

fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java

@@ -9446,7 +9446,12 @@ public BranchOptions visitBranchOptions(DorisParser.BranchOptionsContext ctx) {
         if (ctx.retainTime() != null) {
             DorisParser.TimeValueWithUnitContext time = ctx.retainTime().timeValueWithUnit();
             if (time != null) {
-                retainTime = Optional.of(visitTimeValueWithUnit(time));
+                long retainTimeMs = visitTimeValueWithUnit(time);
+                if (retainTimeMs <= 0) {
+                    throw new IllegalArgumentException(
+                        "RETAIN time value must be greater than 0, got: " + retainTimeMs + " ms");
+                }
+                retainTime = Optional.of(retainTimeMs);
             }
         }
 
@@ -9455,11 +9460,21 @@ public BranchOptions visitBranchOptions(DorisParser.BranchOptionsContext ctx) {
         if (ctx.retentionSnapshot() != null) {
             DorisParser.RetentionSnapshotContext retentionSnapshotContext = ctx.retentionSnapshot();
             if (retentionSnapshotContext.minSnapshotsToKeep() != null) {
-                numSnapshots = Optional.of(
-                    Integer.parseInt(retentionSnapshotContext.minSnapshotsToKeep().value.getText()));
+                int snapshotsCount = Integer.parseInt(
+                        retentionSnapshotContext.minSnapshotsToKeep().value.getText());
+                if (snapshotsCount <= 0) {
+                    throw new IllegalArgumentException(
+                        "Snapshot count (SNAPSHOTS) value must be greater than 0, got: " + snapshotsCount);
+                }
+                numSnapshots = Optional.of(snapshotsCount);
             }
             if (retentionSnapshotContext.timeValueWithUnit() != null) {
-                retention = Optional.of(visitTimeValueWithUnit(retentionSnapshotContext.timeValueWithUnit()));
+                long retentionMs = visitTimeValueWithUnit(retentionSnapshotContext.timeValueWithUnit());
+                if (retentionMs <= 0) {
+                    throw new IllegalArgumentException(
+                        "Retention time value must be greater than 0, got: " + retentionMs + " ms");
+                }
+                retention = Optional.of(retentionMs);
             }
         }
         return new BranchOptions(snapshotId, retainTime, numSnapshots, retention);
@@ -9497,7 +9512,12 @@ public TagOptions visitTagOptions(DorisParser.TagOptionsContext ctx) {
         if (ctx.retainTime() != null) {
             DorisParser.TimeValueWithUnitContext time = ctx.retainTime().timeValueWithUnit();
             if (time != null) {
-                retainTime = Optional.of(visitTimeValueWithUnit(time));
+                long retainTimeMs = visitTimeValueWithUnit(time);
+                if (retainTimeMs <= 0) {
+                    throw new IllegalArgumentException(
+                        "RETAIN time value must be greater than 0, got: " + retainTimeMs + " ms");
+                }
+                retainTime = Optional.of(retainTimeMs);
             }
         }
 

fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/BindRelation.java

@@ -417,6 +417,7 @@ private LogicalPlan getLogicalPlan(TableIf table, UnboundRelation unboundRelatio
 
         List<String> qualifierWithoutTableName = qualifiedTableName.subList(0, qualifiedTableName.size() - 1);
         cascadesContext.getStatementContext().loadSnapshots(
+                table,
                 unboundRelation.getTableSnapshot(),
                 Optional.ofNullable(unboundRelation.getScanParams()));
         boolean isView = false;

fe/fe-core/src/test/java/org/apache/doris/nereids/trees/plans/commands/info/CreateOrReplaceBranchOrTagInfoTest.java

@@ -60,15 +60,6 @@ public void testCreateBranchIfNotExistsToSql() {
         Assertions.assertEquals(expected, branchInfo.toSql());
     }
 
-    @Test
-    public void testCreateOrReplaceBranchIfNotExistsToSql() {
-        // Test CREATE OR REPLACE BRANCH IF NOT EXISTS
-        CreateOrReplaceBranchInfo branchInfo = new CreateOrReplaceBranchInfo(
-                "test_branch", true, true, true, null);
-        String expected = "CREATE OR REPLACE BRANCH IF NOT EXISTS test_branch";
-        Assertions.assertEquals(expected, branchInfo.toSql());
-    }
-
     @Test
     public void testCreateBranchWithSnapshotIdToSql() {
         // Test CREATE BRANCH with snapshot ID

regression-test/data/external_table_p0/iceberg/branch_tag/iceberg_branch_complex_queries.out

@@ -0,0 +1,125 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !agg_count --
+8
+
+-- !agg_sum --
+360
+
+-- !agg_avg --
+45
+
+-- !agg_max --
+80
+
+-- !agg_min --
+10
+
+-- !agg_group_by --
+cat1	3	100
+cat2	3	150
+cat3	2	110
+
+-- !window_row_number --
+1	a	10	1
+2	b	20	2
+3	c	30	3
+4	d	40	4
+5	e	50	5
+6	f	60	6
+7	g	70	7
+8	h	80	8
+
+-- !window_rank --
+1	a	10	1
+2	b	20	2
+3	c	30	1
+4	d	40	2
+5	e	50	1
+6	f	60	2
+7	g	70	3
+8	h	80	3
+
+-- !window_dense_rank --
+1	a	10	1
+2	b	20	2
+3	c	30	3
+4	d	40	4
+5	e	50	5
+6	f	60	6
+7	g	70	7
+8	h	80	8
+
+-- !join_branch_branch --
+1	a	info1
+2	b	info2
+3	c	info3
+6	f	info6
+7	g	info7
+8	h	info8
+
+-- !join_branch_main --
+1	a	info1
+2	b	info2
+3	c	info3
+6	f	info6
+7	g	info7
+
+-- !join_main_branch --
+1	a	info1
+2	b	info2
+3	c	info3
+
+-- !subquery_in --
+
+-- !subquery_exists --
+
+-- !subquery_scalar --
+1	a	80
+2	b	80
+3	c	80
+4	d	80
+5	e	80
+
+-- !cte_simple --
+6	f	60	cat3
+7	g	70	cat1
+8	h	80	cat2
+
+-- !cte_multiple --
+1	a	10	cat1
+2	b	20	cat1
+3	c	30	cat2
+4	d	40	cat2
+7	g	70	cat1
+8	h	80	cat2
+
+-- !cte_join --
+1	a	10
+2	b	20
+3	c	30
+4	d	40
+5	e	50
+6	f	\N
+7	g	\N
+8	h	\N
+
+-- !complex_filter --
+cat1	3	33.33333333333334
+cat2	3	50
+cat3	1	60
+
+-- !order_limit --
+8	h	80	cat2
+7	g	70	cat1
+6	f	60	cat3
+
+-- !order_by_multiple --
+7	g	70	cat1
+2	b	20	cat1
+1	a	10	cat1
+8	h	80	cat2
+4	d	40	cat2
+3	c	30	cat2
+6	f	60	cat3
+5	e	50	cat3
+

regression-test/data/external_table_p0/iceberg/branch_tag/iceberg_branch_cross_operations.out

@@ -0,0 +1,104 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !b1_initial --
+1	a
+2	b
+3	c
+4	d
+5	e
+
+-- !b2_initial --
+1	a
+2	b
+3	c
+
+-- !b2_after_copy --
+1	a
+1	a
+2	b
+2	b
+3	c
+3	c
+4	d
+5	e
+
+-- !b1_unchanged --
+1	a
+2	b
+3	c
+4	d
+5	e
+
+-- !b3_has_main_data --
+1	a
+1	a
+2	b
+2	b
+3	c
+3	c
+6	f
+6	f
+7	g
+7	g
+
+-- !main_unchanged_after_copy --
+1	a
+2	b
+3	c
+6	f
+7	g
+
+-- !main_overwritten --
+1	a
+2	b
+3	c
+6	f
+7	g
+8	h
+9	i
+10	j
+
+-- !b4_unchanged --
+1	a
+2	b
+3	c
+6	f
+7	g
+8	h
+9	i
+10	j
+
+-- !b4_has_filtered_data --
+1	a
+2	b
+3	c
+6	f
+7	g
+8	h
+9	i
+10	j
+11	k
+12	l
+13	m
+
+-- !b6_overwritten --
+1	a
+2	b
+3	c
+6	f
+7	g
+8	h
+9	i
+10	j
+11	k
+12	l
+13	m
+
+-- !b5_chain --
+31
+
+-- !b6_chain --
+20
+
+-- !b7_chain --
+9
+