[Feature] support create or replace pipe

[mofeiatwork]

Why I'm doing:

What I'm doing:

• Support syntax CREATE OR REPLACE PIPE

Fixes #37509

What type of PR is this:

• BugFix
• Feature
• Enhancement
• Refactor
• UT
• Doc
• Tool

Does this PR entail a change in behavior?

• Yes, this PR will result in a change in behavior.
• No, this PR will not result in a change in behavior.

If yes, please specify the type of change:

• Interface/UI changes: syntax, type conversion, expression evaluation, display information
• Parameter changes: default values, similar parameters but with different default values
• Policy changes: use new policy to replace old one, functionality automatically enabled
• Feature removed
• Miscellaneous: upgrade & downgrade compatibility, etc.

Checklist:

• I have added test cases for my bug fix or my new feature
• This pr needs user documentation (for new or modified features or behaviors)
  • I have added documentation for my new feature or new function
• This is a backport pr

Bugfix cherry-pick branch check:

• I have checked the version labels which the pr will be auto-backported to the target branch
  • 3.2
  • 3.1
  • 3.0
  • 2.5

[starrocks-cr]

The most risky bug in this code is:
Attempting to replace a Pipe may fail if pipeMap.get(nameToId.get(dbIdAndName)) returns null, leading to a NullPointerException.

You can modify the code like this:

@@ -74,8 +74,13 @@
                 } else if (stmt.isReplace()) {
                     LOG.info("Pipe {} already exist, replace it with a new one", stmt.getPipeName());
                     Long pipeId = nameToId.get(dbIdAndName.getFirst());
-                    Pipe pipe = pipeMap.get(pipeId);
+                    Pipe pipe = pipeMap.get(dbIdAndName.getFirst());
+                    if (pipe == null) {
+                        throw new DdlException("Cannot replace non-existing pipe with name: " + dbIdAndName.getSecond());
+                    }
                     dropPipeImpl(pipe);
+                }
             }
 
             // Add pipe

In the modification above, I've added a null check for pipe before attempting to replace it. If pipe is null, an exception is thrown indicating that the pipe cannot be replaced because it doesn't exist. This avoids the potential NullPointerException. Additionally, make sure to use the correct key when obtaining pipe from pipeMap; the key should be derived from the resolved dbIdAndName, specifically the first element (dbId) in case dbIdAndName is a pair of database ID and name as the context suggests.

[sonarcloud]

Quality Gate passed

The SonarCloud Quality Gate passed, but some issues were introduced.

4 New issues
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarCloud

[github-actions]

[FE Incremental Coverage Report]

✅ pass : 15 / 15 (100.00%)

file detail

	path	covered_line	new_line	coverage	not_covered_line_detail
🔵	com/starrocks/load/pipe/PipeManager.java	13	13	100.00%	[]
🔵	com/starrocks/sql/ast/pipe/CreatePipeStmt.java	2	2	100.00%	[]

[github-actions]

[BE Incremental Coverage Report]

✅ pass : 0 / 0 (0%)

[mofeiatwork]

@Mergifyio backport branch-3.2

[mergify]

backport branch-3.2

✅ Backports have been created

• #37698 [Feature] support create or replace pipe (backport #37658) has been created for branch branch-3.2